matrix:
allow_failures:
- - python: "3.4"
- python: "pypy"
Programming with the Twitter api classes
========================================
-
The Twitter and TwitterStream classes are the key to building your own
Twitter-enabled applications.
**[http://dev.twitter.com/doc](http://dev.twitter.com/doc)**
-
-Examples::
-
+Examples:
```python
from twitter import *
-# see "Authentication" section below for tokens and keys
t = Twitter(
- auth=OAuth(OAUTH_TOKEN, OAUTH_SECRET,
- CONSUMER_KEY, CONSUMER_SECRET)
- )
+ auth=OAuth(token, token_key, con_secret, con_secret_key)))
# Get your "home" timeline
t.statuses.home_timeline()
t.user.list.members(user="tamtar", list="things-that-are-rad")
# An *optional* `_timeout` parameter can also be used for API
-# calls which take much more time than normal or Twitter stops
-# responding for some reason
-t.users.lookup(screen_name=','.join(A_LIST_OF_100_SCREEN_NAMES), _timeout=1)
+# calls which take much more time than normal or twitter stops
+# responding for some reason:
+t.users.lookup(
+ screen_name=','.join(A_LIST_OF_100_SCREEN_NAMES), _timeout=1)
# Overriding Method: GET/POST
# you should not need to use this method as this library properly
# detects whether GET or POST should be used, Nevertheless
# to force a particular method, use `_method`
t.statuses.oembed(_id=1234567890, _method='GET')
+
+# Send a tweet with an image included (or set your banner or logo similarily)
+# by just reading your image from the web or a file in a string:
+with open("example.png", "rb") as imagefile:
+ params = {"media[]": imagefile.read(), "status": "PTT"}
+t.statuses.update_with_media(**params)
+
+# Or by sending a base64 encoded image:
+params = {"media[]": base64_image, "status": "PTT", "_base64": True}
+t.statuses.update_with_media(**params)
```
-Searching Twitter::
-``` python
+Searching Twitter:
+```python
# Search for the latest tweets about #pycon
t.search.tweets(q="#pycon")
```
-----------------------
Twitter API calls return decoded JSON. This is converted into
-a bunch of Python lists, dicts, ints, and strings. For example::
+a bunch of Python lists, dicts, ints, and strings. For example:
```python
x = twitter.statuses.home_timeline()
--------------------
If you prefer to get your Twitter data in XML format, pass
-format="xml" to the Twitter object when you instantiate it::
+format="xml" to the Twitter object when you instantiate it:
```python
twitter = Twitter(format="xml")
The TwitterStream class
-----------------------
-The TwitterStream object is an interface to the Twitter Stream API
-(stream.twitter.com). This can be used pretty much the same as the
-Twitter class except the result of calling a method will be an
-iterator that yields objects decoded from the stream. For
-example::
+The TwitterStream object is an interface to the Twitter Stream
+API. This can be used pretty much the same as the Twitter class
+except the result of calling a method will be an iterator that
+yields objects decoded from the stream. For example::
```python
-twitter_stream = TwitterStream(auth=UserPassAuth('joe', 'joespassword'))
+twitter_stream = TwitterStream(auth=OAuth(...))
iterator = twitter_stream.statuses.sample()
for tweet in iterator:
- # ...do something with this tweet...
+ ...do something with this tweet...
```
-The iterator will yield tweets forever and ever (until the stream
-breaks at which point it raises a TwitterHTTPError.)
-
-The `block` parameter controls if the stream is blocking. Default
-is blocking (True). When set to False, the iterator will
-occasionally yield None when there is no available message.
-
Per default the ``TwitterStream`` object uses
[public streams](https://dev.twitter.com/docs/streaming-apis/streams/public).
If you want to use one of the other
print msg['direct_message']['text']
```
+The iterator will yield until the TCP connection breaks. When the
+connection breaks, the iterator yields `{'hangup': True}`, and
+raises `StopIteration` if iterated again.
+
+Similarly, if the stream does not produce heartbeats for more than
+90 seconds, the iterator yields `{'hangup': True,
+'heartbeat_timeout': True}`, and raises `StopIteration` if
+iterated again.
+
+The `timeout` parameter controls the maximum time between
+yields. If it is nonzero, then the iterator will yield either
+stream data or `{'timeout': True}` within the timeout period. This
+is useful if you want your program to do other stuff in between
+waiting for tweets.
+
+The `block` parameter sets the stream to be fully non-blocking. In
+this mode, the iterator always yields immediately. It returns
+stream data, or `None`. Note that `timeout` supercedes this
+argument, so it should also be set `None` to use this mode.
+
Twitter Response Objects
------------------------
-Response from a Twitter request. Behaves like a list or a string
+Response from a twitter request. Behaves like a list or a string
(depending on requested format) but it has a few other interesting
attributes.
`headers` gives you access to the response headers as an
httplib.HTTPHeaders instance. You can do
-`response.headers.getheader('h')` to retrieve a header.
+`response.headers.get('h')` to retrieve a header.
Authentication
--------------
You can authenticate with Twitter in three ways: NoAuth, OAuth, or
-UserPassAuth. Get help() on these classes to learn how to use them.
+OAuth2 (app-only). Get help() on these classes to learn how to use them.
-OAuth is probably the most useful.
+OAuth and OAuth2 are probably the most useful.
Working with OAuth
This will get you a CONSUMER_KEY and CONSUMER_SECRET.
When users run your application they have to authenticate your app
-with their Twitter account. A few HTTP calls to Twitter are required
+with their Twitter account. A few HTTP calls to twitter are required
to do this. Please see the twitter.oauth_dance module to see how this
is done. If you are making a command-line app, you can use the
oauth_dance() function directly.
-Performing the "oauth dance" gets you an oauth token and oauth secret
+Performing the "oauth dance" gets you an ouath token and oauth secret
that authenticate the user with Twitter. You should save these for
later so that the user doesn't have to do the oauth dance again.
strings in the file. Not terribly exciting.
Finally, you can use the OAuth authenticator to connect to Twitter. In
-code it all goes like this::
+code it all goes like this:
```python
from twitter import *
oauth_token, oauth_secret = read_token_file(MY_TWITTER_CREDS)
twitter = Twitter(auth=OAuth(
- oauth_token, oauth_secret, CONSUMER_KEY, CONSUMER_SECRET))
+ oauth_token, oauth_token_secret, CONSUMER_KEY, CONSUMER_SECRET))
# Now work with Twitter
twitter.statuses.update(status='Hello, world!')
```
+Working with OAuth2
+-------------------
+
+Twitter only supports the application-only flow of OAuth2 for certain
+API endpoints. This OAuth2 authenticator only supports the application-only
+flow right now.
+
+To authenticate with OAuth2, visit the Twitter developer page and create a new
+application:
+
+**[https://dev.twitter.com/apps/new](https://dev.twitter.com/apps/new)**
+
+This will get you a CONSUMER_KEY and CONSUMER_SECRET.
+
+Exchange your CONSUMER_KEY and CONSUMER_SECRET for a bearer token using the
+oauth2_dance function.
+
+Finally, you can use the OAuth2 authenticator and your bearer token to connect
+to Twitter. In code it goes like this::
+
+```python
+twitter = Twitter(auth=OAuth2(bearer_token=BEARER_TOKEN))
+
+# Now work with Twitter
+twitter.search.tweets(q='keyword')
+```
License
=======
from setuptools import setup, find_packages
import sys, os
-version = '1.14.3'
+version = '1.15.0'
install_requires = [
# -*- Extra requirements: -*-
]
+if sys.version_info < (2,7):
+ install_requires.append('argparse')
setup(name='twitter',
version=version,
from .api import Twitter, TwitterError, TwitterHTTPError, TwitterResponse
from .auth import NoAuth, UserPassAuth
-from .oauth import (OAuth, read_token_file, write_token_file,
- __doc__ as oauth_doc)
-from .oauth2 import OAuth2
+from .oauth import (
+ OAuth, read_token_file, write_token_file,
+ __doc__ as oauth_doc)
+from .oauth2 import (
+ OAuth2, read_bearer_token_file, write_bearer_token_file,
+ __doc__ as oauth2_doc)
from .stream import TwitterStream
-from .oauth_dance import oauth_dance
+from .oauth_dance import oauth_dance, oauth2_dance
__doc__ = __doc__ or ""
--------------
You can authenticate with Twitter in three ways: NoAuth, OAuth, or
-UserPassAuth. Get help() on these classes to learn how to use them.
+OAuth2 (app-only). Get help() on these classes to learn how to use them.
-OAuth is probably the most useful.
+OAuth and OAuth2 are probably the most useful.
Working with OAuth
__doc__ += dedent(oauth_doc or "")
-__all__ = ["Twitter", "TwitterStream", "TwitterResponse", "TwitterError",
- "TwitterHTTPError", "NoAuth", "OAuth", "UserPassAuth",
- "read_token_file", "write_token_file", "oauth_dance", "OAuth2"]
+__doc__ += """
+Working with OAuth2
+-------------------
+"""
+
+__doc__ += dedent(oauth2_doc or "")
+
+__all__ = [
+ "NoAuth",
+ "OAuth",
+ "OAuth2",
+ "oauth2_dance",
+ "oauth_dance",
+ "read_bearer_token_file",
+ "read_token_file",
+ "Twitter",
+ "TwitterError",
+ "TwitterHTTPError",
+ "TwitterResponse",
+ "TwitterStream",
+ "UserPassAuth",
+ "write_bearer_token_file",
+ "write_token_file",
+ ]
from .auth import NoAuth
import re
+import sys
import gzip
try:
res = response
return res
+def method_for_uri(uri):
+ method = "GET"
+ for action in POST_ACTIONS:
+ if re.search("%s(/\d+)?$" % action, uri):
+ method = "POST"
+ break
+ return method
class TwitterCall(object):
uriparts.append(str(kwargs.pop(uripart, uripart)))
uri = '/'.join(uriparts)
- method = kwargs.pop('_method', None)
- if not method:
- method = "GET"
- for action in POST_ACTIONS:
- if re.search("%s(/\d+)?$" % action, uri):
- method = "POST"
- break
+ method = kwargs.pop('_method', None) or method_for_uri(uri)
# If an id kwarg is present and there is no id to fill in in
# the list of uriparts, assume the id goes at the end.
uriBase = "http%s://%s/%s%s%s" % (
secure_str, self.domain, uri, dot, self.format)
+ # Check if argument tells whether img is already base64 encoded
+ b64_convert = True
+ if "_base64" in kwargs:
+ b64_convert = not kwargs.pop("_base64")
+ if b64_convert:
+ import base64
+
# Catch media arguments to handle oauth query differently for multipart
media = None
- for arg in ['media[]', 'banner', 'image']:
+ for arg in ['media[]']:
if arg in kwargs:
media = kwargs.pop(arg)
+ if b64_convert:
+ media = base64.b64encode(media)
+ if sys.version_info >= (3, 0):
+ media = str(media, 'utf8')
mediafield = arg
break
+ # Catch media arguments that are not accepted through multipart
+ # and are not yet base64 encoded
+ if b64_convert:
+ for arg in ['banner', 'image']:
+ if arg in kwargs:
+ kwargs[arg] = base64.b64encode(kwargs[arg])
+
headers = {'Accept-Encoding': 'gzip'} if self.gzip else dict()
body = None
arg_data = None
bod.append('--' + BOUNDARY)
bod.append(
'Content-Disposition: form-data; name="%s"' % mediafield)
+ bod.append('Content-Transfer-Encoding: base64')
bod.append('')
bod.append(media)
for k, v in kwargs.items():
bod.append('')
bod.append(v)
bod.append('--' + BOUNDARY + '--')
- body = '\r\n'.join(bod)
+ body = '\r\n'.join(bod).encode('utf8')
headers['Content-Type'] = \
'multipart/form-data; boundary=%s' % BOUNDARY
buf = StringIO(data)
f = gzip.GzipFile(fileobj=buf)
data = f.read()
- if "json" == self.format:
+ if len(data) == 0:
+ return wrap_response({}, handle.headers)
+ elif "json" == self.format:
res = json.loads(data.decode('utf8'))
return wrap_response(res, handle.headers)
else:
Examples::
+ from twitter import *
+
t = Twitter(
auth=OAuth(token, token_key, con_secret, con_secret_key)))
# Get your "home" timeline
t.statuses.home_timeline()
- # Get a particular friend's tweets
- t.statuses.user_timeline(user_id="billybob")
+ # Get a particular friend's timeline
+ t.statuses.user_timeline(screen_name="billybob")
+
+ # to pass in GET/POST parameters, such as `count`
+ t.statuses.home_timeline(count=5)
+
+ # to pass in the GET/POST parameter `id` you need to use `_id`
+ t.statuses.oembed(_id=1234567890)
# Update your status
t.statuses.update(
# An *optional* `_timeout` parameter can also be used for API
# calls which take much more time than normal or twitter stops
- # responding for some reasone
+ # responding for some reason:
t.users.lookup(
screen_name=','.join(A_LIST_OF_100_SCREEN_NAMES), \
_timeout=1)
+ # Overriding Method: GET/POST
+ # you should not need to use this method as this library properly
+ # detects whether GET or POST should be used, Nevertheless
+ # to force a particular method, use `_method`
+ t.statuses.oembed(_id=1234567890, _method='GET')
+
+ # Send a tweet with an image included (or set your banner or logo similarily)
+ # by just reading your image from the web or a file in a string:
+ with open("example.png", "rb") as imagefile:
+ params = {"media[]": imagefile.read(), "status": "PTT"}
+ t.statuses.update_with_media(**params)
+
+ # Or by sending a base64 encoded image:
+ params = {"media[]": base64_image, "status": "PTT", "_base64": True}
+ t.statuses.update_with_media(**params)
Searching Twitter::
# authenticate using OAuth, asking for token if necessary
if options['oauth']:
- oauth_filename = (os.getenv("HOME", "") + os.sep
- + ".twitter-archiver_oauth")
+ oauth_filename = (os.environ.get('HOME',
+ os.environ.get('USERPROFILE', ''))
+ + os.sep
+ + '.twitter-archiver_oauth')
+
if not os.path.exists(oauth_filename):
oauth_dance("Twitter-Archiver", CONSUMER_KEY, CONSUMER_SECRET,
oauth_filename)
printNicely(lf(list))
return []
else:
- return reversed(twitter.user.lists.list.statuses(
- user=screen_name, list=options['extra_args'][1]))
+ return reversed(twitter.lists.statuses(
+ owner_screen_name=screen_name, slug=options['extra_args'][1]))
class MyListsAction(ListsAction):
Finally, you can use the OAuth authenticator to connect to Twitter. In
code it all goes like this::
+ from twitter import *
+
MY_TWITTER_CREDS = os.path.expanduser('~/.my_app_credentials')
if not os.path.exists(MY_TWITTER_CREDS):
oauth_dance("My App Name", CONSUMER_KEY, CONSUMER_SECRET,
"""
-Visit the Twitter developer page and create a new application:
+Twitter only supports the application-only flow of OAuth2 for certain
+API endpoints. This OAuth2 authenticator only supports the application-only
+flow right now.
+
+To authenticate with OAuth2, visit the Twitter developer page and create a new
+application:
https://dev.twitter.com/apps/new
This will get you a CONSUMER_KEY and CONSUMER_SECRET.
-Twitter only supports the application-only flow of OAuth2 for certain
-API endpoints. This OAuth2 authenticator only supports the application-only
-flow right now. If twitter supports OAuth2 for other endpoints, this
-authenticator may be modified as needed.
+Exchange your CONSUMER_KEY and CONSUMER_SECRET for a bearer token using the
+oauth2_dance function.
-Finally, you can use the OAuth2 authenticator to connect to Twitter. In
-code it all goes like this::
+Finally, you can use the OAuth2 authenticator and your bearer token to connect
+to Twitter. In code it goes like this::
twitter = Twitter(auth=OAuth2(bearer_token=BEARER_TOKEN))
from base64 import b64encode
from .auth import Auth
+def write_bearer_token_file(filename, oauth2_bearer_token):
+ """
+ Write a token file to hold the oauth2 bearer token.
+ """
+ oauth_file = open(filename, 'w')
+ print(oauth2_bearer_token, file=oauth_file)
+ oauth_file.close()
+
+def read_bearer_token_file(filename):
+ """
+ Read a token file and return the oauth2 bearer token.
+ """
+ f = open(filename)
+ return f.readline().strip()
class OAuth2(Auth):
"""
consumer_secret if you are requesting a bearer_token. Otherwise
you must supply the bearer_token.
"""
- self.bearer_token = None
- self.consumer_key = None
- self.consumer_secret = None
-
- if bearer_token:
- self.bearer_token = bearer_token
- elif consumer_key and consumer_secret:
- self.consumer_key = consumer_key
- self.consumer_secret = consumer_secret
- else:
+ self.bearer_token = bearer_token
+ self.consumer_key = consumer_key
+ self.consumer_secret = consumer_secret
+
+ if not (bearer_token or (consumer_key and consumer_secret)):
raise MissingCredentialsError(
'You must supply either a bearer token, or both a '
'consumer_key and a consumer_secret.')
def encode_params(self, base_url, method, params):
-
return urlencode(params)
def generate_headers(self):
b'Authorization': 'Bearer {0}'.format(
self.bearer_token).encode('utf8')
}
-
- elif self.consumer_key and self.consumer_secret:
-
+ else:
headers = {
b'Content-Type': (b'application/x-www-form-urlencoded;'
b'charset=UTF-8'),
).decode('utf8')
).encode('utf8')
}
-
- else:
- raise MissingCredentialsError(
- 'You must supply either a bearer token, or both a '
- 'consumer_key and a consumer_secret.')
-
return headers
import webbrowser
import time
-from .api import Twitter
+from .api import Twitter, json
from .oauth import OAuth, write_token_file
+from .oauth2 import OAuth2, write_bearer_token_file
try:
_input = raw_input
_input = input
+def oauth2_dance(consumer_key, consumer_secret, token_filename=None):
+ """
+ Perform the OAuth2 dance to transform a consumer key and secret into a
+ bearer token.
+
+ If a token_filename is given, the bearer token will be written to
+ the file.
+ """
+ twitter = Twitter(
+ auth=OAuth2(consumer_key=consumer_key, consumer_secret=consumer_secret),
+ format="",
+ api_version="")
+ token = json.loads(twitter.oauth2.token(grant_type="client_credentials")
+ .encode("utf8"))["access_token"]
+ if token_filename:
+ write_bearer_token_file(token)
+ return token
def oauth_dance(app_name, consumer_key, consumer_secret, token_filename=None):
"""
auth=OAuth('', '', consumer_key, consumer_secret),
format='', api_version=None)
oauth_token, oauth_token_secret = parse_oauth_tokens(
- twitter.oauth.request_token())
+ twitter.oauth.request_token(oauth_callback="oob"))
print("""
In the web browser window that opens please choose to Allow
access. Copy the PIN number that appears on the next page and paste or
iterator = twitter_stream.statuses.sample()
for tweet in iterator:
- ...do something with this tweet...
+ # ...do something with this tweet...
+
+ Per default the ``TwitterStream`` object uses
+ [public streams](https://dev.twitter.com/docs/streaming-apis/streams/public).
+ If you want to use one of the other
+ [streaming APIs](https://dev.twitter.com/docs/streaming-apis), specify the URL
+ manually:
+
+ - [Public streams](https://dev.twitter.com/docs/streaming-apis/streams/public): stream.twitter.com
+ - [User streams](https://dev.twitter.com/docs/streaming-apis/streams/user): userstream.twitter.com
+ - [Site streams](https://dev.twitter.com/docs/streaming-apis/streams/site): sitestream.twitter.com
+
+ Note that you require the proper
+ [permissions](https://dev.twitter.com/docs/application-permission-model) to
+ access these streams. E.g. for direct messages your
+ [application](https://dev.twitter.com/apps) needs the "Read, Write & Direct
+ Messages" permission.
+
+ The following example demonstrates how to retrieve all new direct messages
+ from the user stream::
+
+ auth = OAuth(
+ consumer_key='[your consumer key]',
+ consumer_secret='[your consumer secret]',
+ token='[your token]',
+ token_secret='[your token secret]'
+ )
+ twitter_userstream = TwitterStream(auth=auth, domain='userstream.twitter.com')
+ for msg in twitter_userstream.user():
+ if 'direct_message' in msg:
+ print msg['direct_message']['text']
The iterator will yield until the TCP connection breaks. When the
connection breaks, the iterator yields `{'hangup': True}`, and
from twitter.stream import TwitterStream, Timeout, HeartbeatTimeout, Hangup
from twitter.oauth import OAuth
+from twitter.oauth2 import OAuth2, read_bearer_token_file
from twitter.util import printNicely
-
def parse_arguments():
parser = argparse.ArgumentParser(description=__doc__ or "")
POST_ACTIONS = [
# Status Methods
- 'update', 'retweet', 'update_with_media',
+ 'update', 'retweet', 'update_with_media', 'statuses/lookup',
# Direct Message Methods
'new',
'create', 'create_all',
# Users Methods
- 'lookup', 'report_spam',
+ 'users/lookup', 'report_spam',
# Streaming Methods
'filter', 'user', 'site',
projects / z_archive / twitter.git / commitdiff
