Library for accessing the X API (Twitter)
npx @tessl/cli install tessl/pypi-tweepy@4.16.0A comprehensive Python library for accessing the X API (Twitter), providing both synchronous and asynchronous interfaces to Twitter's API v1.1 (legacy) and API v2. Tweepy enables developers to build applications that interact with X/Twitter's social media platform, offering complete coverage of API endpoints for posting tweets, managing user accounts, searching content, streaming real-time data, and handling media uploads.
pip install tweepypip install tweepy[async] for async supportimport tweepyCommon imports for different functionality:
# Twitter API v2 Client
from tweepy import Client
# Authentication handlers
from tweepy import OAuth1UserHandler, OAuth2BearerHandler, OAuth2UserHandler
# Streaming
from tweepy import StreamingClient, StreamRule
# Data models
from tweepy import Tweet, User, Media, Space, List
# Legacy API v1.1
from tweepy import API
# Pagination utilities
from tweepy import Cursor, Paginatorimport tweepy
# Initialize client with Bearer token (App-only auth)
client = tweepy.Client(bearer_token="your_bearer_token")
# Search for recent tweets
tweets = client.search_recent_tweets(query="python programming", max_results=10)
# Print tweet text and author info
for tweet in tweets.data:
print(f"@{tweet.author_id}: {tweet.text}")
# Get user information
user = client.get_user(username="twitter")
print(f"User: {user.data.name} (@{user.data.username})")
print(f"Followers: {user.data.public_metrics['followers_count']}")import tweepy
# OAuth 1.0a User Context authentication
auth = tweepy.OAuth1UserHandler(
consumer_key="your_consumer_key",
consumer_secret="your_consumer_secret",
access_token="your_access_token",
access_token_secret="your_access_token_secret"
)
# Create client with user authentication
client = tweepy.Client(
consumer_key="your_consumer_key",
consumer_secret="your_consumer_secret",
access_token="your_access_token",
access_token_secret="your_access_token_secret"
)
# Post a tweet (requires user auth)
response = client.create_tweet(text="Hello Twitter from Tweepy!")
print(f"Tweet posted: {response.data['id']}")import tweepy
class TweetPrinter(tweepy.StreamingClient):
def on_tweet(self, tweet):
print(f"New tweet: {tweet.text}")
# Initialize streaming client
stream = TweetPrinter(bearer_token="your_bearer_token")
# Add filter rules
rule = tweepy.StreamRule("python -is:retweet lang:en")
stream.add_rules(rule)
# Start streaming (blocking call)
stream.filter()Tweepy provides multiple interfaces for different use cases:
Comprehensive interface to Twitter API v2 endpoints including tweet management, user operations, timelines, search, lists, spaces, direct messages, and media upload. Supports both app-only and user authentication contexts.
class Client:
def __init__(self, bearer_token=None, consumer_key=None, consumer_secret=None,
access_token=None, access_token_secret=None, *,
return_type=Response, wait_on_rate_limit=False): ...
# Tweet operations
def create_tweet(self, text=None, **kwargs): ...
def delete_tweet(self, id, **kwargs): ...
def get_tweet(self, id, **kwargs): ...
def search_recent_tweets(self, query, **kwargs): ...
# User operations
def get_me(self, **kwargs): ...
def get_user(self, *, id=None, username=None, **kwargs): ...
def follow_user(self, target_user_id, **kwargs): ...
# Streaming and more...Multiple OAuth authentication handlers supporting different application types and use cases, from simple app-only access to full user authorization flows.
class OAuth1UserHandler:
def __init__(self, consumer_key, consumer_secret, access_token=None,
access_token_secret=None, callback=None): ...
def get_authorization_url(self, signin_with_twitter=False, access_type=None): ...
def get_access_token(self, verifier=None): ...
class OAuth2BearerHandler:
def __init__(self, bearer_token): ...
class OAuth2UserHandler:
def __init__(self, *, client_id, redirect_uri, scope, client_secret=None): ...
def get_authorization_url(self): ...
def fetch_token(self, authorization_response): ...Rule-based streaming client for receiving real-time tweets matching specified criteria, with customizable event handlers and automatic reconnection.
class StreamingClient:
def __init__(self, bearer_token, *, chunk_size=512, daemon=False,
max_retries=float('inf'), **kwargs): ...
def filter(self, *, threaded=False, **kwargs): ...
def sample(self, *, threaded=False, **kwargs): ...
def add_rules(self, add, **kwargs): ...
def delete_rules(self, ids, **kwargs): ...
def get_rules(self, **kwargs): ...
# Event handlers (override these)
def on_tweet(self, tweet): ...
def on_data(self, raw_data): ...Rich data model classes representing Twitter objects with full field access and type safety, including tweets, users, media, spaces, lists, and more.
class Tweet:
# Core attributes
id: str
text: str
author_id: str
created_at: datetime
public_metrics: dict
# ... all other v2 tweet fields
class User:
id: str
name: str
username: str
created_at: datetime
public_metrics: dict
# ... all other v2 user fields
class Media:
media_key: str
type: str
url: str
# ... all other media fieldsComplete interface to Twitter API v1.1 endpoints for backward compatibility and access to v1.1-specific functionality not yet available in v2.
class API:
def __init__(self, auth=None, *, cache=None, host='api.twitter.com',
parser=None, **kwargs): ...
# Timeline methods
def home_timeline(self, **kwargs): ...
def user_timeline(self, **kwargs): ...
# Tweet methods
def update_status(self, status, **kwargs): ...
def get_status(self, id, **kwargs): ...
# User methods
def get_user(self, **kwargs): ...
def search_users(self, q, **kwargs): ...
# Many more v1.1 methods...Utility classes for pagination, caching, error handling, and other common functionality to simplify API usage and improve application performance.
class Paginator:
def __init__(self, method, *args, **kwargs): ...
def flatten(self, limit=None): ...
def get_next(self): ...
def get_previous(self): ...
class Cursor:
def __init__(self, method, *args, **kwargs): ...
def items(self, limit=None): ...
def pages(self, limit=None): ...
# Caching classes
class MemoryCache: ...
class FileCache: ...Complete async/await interface providing non-blocking access to all Twitter API functionality with identical method signatures to the synchronous interface.
class AsyncClient:
# Identical methods to Client but async
async def create_tweet(self, text=None, **kwargs): ...
async def get_tweet(self, id, **kwargs): ...
async def search_recent_tweets(self, query, **kwargs): ...
class AsyncStreamingClient:
# Identical methods to StreamingClient but async
async def filter(self, **kwargs): ...
async def sample(self, **kwargs): ...from collections import namedtuple
Response = namedtuple("Response", ("data", "includes", "errors", "meta"))
StreamResponse = namedtuple("StreamResponse", ("data",))class TweepyException(Exception):
"""Base Tweepy exception"""
class HTTPException(TweepyException):
"""HTTP request failure exception"""
class BadRequest(HTTPException):
"""400 Bad Request error"""
class Unauthorized(HTTPException):
"""401 Unauthorized error"""
class Forbidden(HTTPException):
"""403 Forbidden error"""
class NotFound(HTTPException):
"""404 Not Found error"""
class TooManyRequests(HTTPException):
"""429 Rate limit exceeded error"""
class TwitterServerError(HTTPException):
"""5xx server errors"""# Tweet field constants
TWEET_FIELDS = (
"id", "text", "attachments", "author_id", "context_annotations",
"conversation_id", "created_at", "edit_controls", "entities", "geo",
"in_reply_to_user_id", "lang", "non_public_metrics", "organic_metrics",
"possibly_sensitive", "promoted_metrics", "public_metrics",
"referenced_tweets", "reply_settings", "source", "withheld"
)
PUBLIC_TWEET_FIELDS = (
"id", "text", "attachments", "author_id", "context_annotations",
"conversation_id", "created_at", "edit_controls", "entities", "geo",
"in_reply_to_user_id", "lang", "possibly_sensitive", "public_metrics",
"referenced_tweets", "reply_settings", "source", "withheld"
)
# User field constants
USER_FIELDS = (
"id", "name", "username", "created_at", "description", "entities",
"location", "pinned_tweet_id", "profile_image_url", "protected",
"public_metrics", "url", "verified", "verified_type", "withheld"
)
# Media field constants
MEDIA_FIELDS = (
"media_key", "type", "url", "duration_ms", "height", "non_public_metrics",
"organic_metrics", "preview_image_url", "promoted_metrics",
"public_metrics", "width", "alt_text", "variants"
)
# Additional field constants for Lists, Spaces, Places, Polls, etc.
LIST_FIELDS = ("id", "name", "created_at", "description", "follower_count", "member_count", "private", "owner_id")
SPACE_FIELDS = ("id", "state", "created_at", "ended_at", "host_ids", "lang", "is_ticketed", "invited_user_ids", "participant_count", "subscriber_count", "scheduled_start", "speaker_ids", "started_at", "title", "topic_ids", "updated_at", "creator_id")
PLACE_FIELDS = ("id", "full_name", "name", "country", "country_code", "geo", "place_type")
POLL_FIELDS = ("id", "options", "duration_minutes", "end_datetime", "voting_status")
DIRECT_MESSAGE_EVENT_FIELDS = ("id", "text", "event_type", "created_at", "sender_id", "dm_conversation_id", "referenced_tweet", "media_keys", "attachments")