Twitter Streaming API with Tweepy
[Updated on 01.2019 ] Tweepy is a Python library for accessing the Twitter API. In the post I will show you how to use Twitter Streaming API and of course, how to get information you want.
Download and install Tweepy
Tweepy is compatible with Python 2.6, 2.7, 3.3 and 3.4
The easiest way to install it is to use pip :
pip install tweepy
Or you can use git :
git clone https://github.com/tweepy/tweepy.git
cd tweepy
python setup.py install
Okay, now let's create a new project.
Authentication
Tweepy handle Authentication, reading messages, following user, etc.
In the following example, I am going to show you how to get the stream and how to display information. If you need more information about how to send messages or how to follow user, you can read Tweepy documentation, which is very well documented: http://docs.tweepy.org/en/latest/
Don't forget to create an account to http://apps.twitter.com and get your consumer key and access token.
Let's create a separated file called : authentication.py with the following code :
class authentication:
def __init__(self):
# Go to http://apps.twitter.com and create an app.
# The consumer key and secret will be generated for you after
self.consumer_key ="xxx"
self.consumer_secret="xxx"
# After the step above, you will be redirected to your app's page.
# Create an access token under the the "Your access token" section
self.access_token="xxx"
self.access_token_secret="xxx"
def getconsumer_key(self):
return self.consumer_key
def getconsumer_secret(self):
return self.consumer_secret
def getaccess_token(self):
return self.access_token
def getaccess_token_secret(self):
return self.access_token_secret
Create a new file called tweepy_example.py and add the following code to connect to Twitter API :
# Add this line to the top
from authentication import authentication
if __name__ == '__main__':
# Get access and key from another class
auth = authentication()
consumer_key = auth.getconsumer_key()
consumer_secret = auth.getconsumer_secret()
access_token = auth.getaccess_token()
access_token_secret = auth.getaccess_token_secret()
# Authentication
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.secure = True
auth.set_access_token(access_token, access_token_secret)
Create a class inheriting from StreamListener
The first thing to do is to implement StreamListener to get the stream.
import tweepy
class TwitterStreamListener(tweepy.StreamListener):
""" A listener handles tweets are the received from the stream.
This is a basic listener that just prints received tweets to stdout.
"""
def on_status(self, status):
get_tweet(status)
get_user_informations(status)
# Twitter error list : https://dev.twitter.com/overview/api/response-codes
def on_error(self, status_code):
if status_code == 403:
print("The request is understood, but it has been refused or access is not allowed. Limit is maybe reached")
return False
on_status(self, status) method receive the tweet (status) and call two methods :
get_tweet which is use to parse the document. get_tweet will be created later.
get_user_informations which is used to get user's informations.
on_error(self, status_code) method is usefull to catch error message
Connect to Twitter Streaming API
# Create a wrapper for the API provided by Twitter
api = tweepy.API(auth)
API wrapper can be overloded with some parameters :
For example if I want to wait when the limit is exceeded and be notified, etc :
api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=10, retry_delay=5, retry_errors=5)
Default parameters :
auth_handler = None // Authentication handler to be used
host = 'api.twitter.com // General API host
search_host = 'search.twitter.com // Search API host
cache = None // Cache backend to use
api_root = '/1' // General API path root
search_root = '' // Search API path root
retry_count = 0 // Default number of retries to attempt when error occurs
retry_delay = 0 // Number of seconds to wait between retries
retry_errors = None // Which HTTP status codes to retry
timeout = 60 // The maximum amount of time to wait for a response from Twitter
parser = ModelParser // The object to use for parsing the response from Twitter
compression = False // Whether or not to use GZIP compression for requests
wait_on_rate_limit = False // Whether or not to automatically wait for rate limits to replenish
wait_on_rate_limit_notify = False // Whether or not to print a notification when Tweepy is waiting for rate limits to replenish
proxy = None // The full url to an HTTPS proxy to use for connecting to Twitter
Source : http://docs.tweepy.org/en/latest/api.html
Create a Stream object
# Create the stream
streamListener = TwitterStreamListener()
# Add your wrapper and your listener to the stream object
myStream = tweepy.Stream(auth=api.auth, listener=streamListener)
# Optional : Filter your tweet by a Hashtag or word
is_async : Stream will run on a new thread
myStream.filter(track=['programming'], is_async=True)
Now you're ready to go! You are receiving tweets and you can play with it by using on_status method of your streamListener.
Let's see how to get particular information such as user's information, tweet message or mention's users.
Get user's informations
I will only print informations. Do whatever you want with it !
def get_user_informations(tweet):
print("User ID \t:" + str(tweet.user.id))
print("User image profil \t:" + tweet.user.profile_image_url_https)
print("User Name \t:" + tweet.user.name)
print("User URL \t:", tweet.user.url)
print("User profil text color \t:" + tweet.user.profile_text_color)
print("User background image url \t:" + tweet.user.profile_background_image_url)
print("User Friends count \t:" + str(tweet.user.friends_count))
print("User Screen name \t:" + tweet.user.screen_name)
print("User Verified \t:" + str(tweet.user.verified))
print("User Favorite count \t:" + str(tweet.user.favourites_count))
if hasattr(tweet.user, 'time_zone'):
print("User Time zone \t:", tweet.user.time_zone)
print("User UTC Offset \t:" + str(tweet.user.utc_offset))
print("User Status count \t:" + str(tweet.user.statuses_count))
print("User Description \t:", tweet.user.description)
print("User Follower count \t:" + str(tweet.user.followers_count))
print("User Created at \t:" + str(tweet.user.created_at))
Get tweet's informations
Display tweet's message, how many times is favorited, if the tweet is favorited by you.
You can also manipulate retweeted message and get mentions users informations !
def get_tweet(tweet):
print("Tweet Message : " + tweet.text)
print("Tweet Favorited \t:" + str(tweet.favorited))
print("Tweet Favorited count \t:" + str(tweet.favorite_count))
# Display sender and mentions user
if hasattr(tweet, 'retweeted_status'):
print("Tweet send by : " + tweet.retweeted_status.user.screen_name)
print("Original tweet ID" + tweet.retweeted_status.id_str)
for screenname in tweet.retweeted_status.entities['user_mentions']:
print("Mention user: " + str(screenname['screen_name']))
Complete code
import tweepy
import socket
import requests
import time
from authentication import authentication # Consumer and access token/key
class TwitterStreamListener(tweepy.StreamListener):
""" A listener handles tweets are the received from the stream.
This is a basic listener that just prints received tweets to stdout.
"""
def on_status(self, status):
get_tweet(status)
get_user_informations(status)
# Twitter error list : https://dev.twitter.com/overview/api/response-codes
def on_error(self, status_code):
if status_code == 403:
print("The request is understood, but it has been refused or access is not allowed. Limit is maybe reached")
return False
def get_tweet(tweet):
print("Tweet Message : " + tweet.text)
print("Tweet Favorited \t:" + str(tweet.favorited))
print("Tweet Favorited count \t:" + str(tweet.favorite_count))
# Display sender and mentions user
if hasattr(tweet, 'retweeted_status'):
print("Tweet send by : " + tweet.retweeted_status.user.screen_name)
print("Original tweet ID" + tweet.retweeted_status.id_str)
for screenname in tweet.retweeted_status.entities['user_mentions']:
print("Mention user: " + str(screenname['screen_name']))
def get_user_informations(tweet):
print("User ID \t:" + str(tweet.user.id))
print("User image profil \t:" + tweet.user.profile_image_url_https)
print("User Name \t:" + tweet.user.name)
print("User URL \t:", tweet.user.url)
print("User profil text color \t:" + tweet.user.profile_text_color)
print("User background image url \t:" + tweet.user.profile_background_image_url)
print("User Friends count \t:" + str(tweet.user.friends_count))
print("User Screen name \t:" + tweet.user.screen_name)
print("User Verified \t:" + str(tweet.user.verified))
print("User Favorite count \t:" + str(tweet.user.favourites_count))
if hasattr(tweet.user, 'time_zone'):
print("User Time zone \t:", tweet.user.time_zone)
print("User UTC Offset \t:" + str(tweet.user.utc_offset))
print("User Status count \t:" + str(tweet.user.statuses_count))
print("User Description \t:", tweet.user.description)
print("User Follower count \t:" + str(tweet.user.followers_count))
print("User Created at \t:" + str(tweet.user.created_at))
if __name__ == '__main__':
# Get access and key from another class
auth = authentication()
consumer_key = auth.getconsumer_key()
consumer_secret = auth.getconsumer_secret()
access_token = auth.getaccess_token()
access_token_secret = auth.getaccess_token_secret()
# Authentication
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.secure = True
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=10, retry_delay=5, retry_errors=5)
streamListener = TwitterStreamListener()
myStream = tweepy.Stream(auth=api.auth, listener=streamListener)
myStream.filter(track=['michael'], is_async=True)
Links
Image credit
"Typography" by Ratha Grimes is licensed under CC BY 2.0