import re
import gzip
+import httplib
try:
import json
except ImportError:
import simplejson as json
+
class _DEFAULT(object):
pass
+
class TwitterError(Exception):
"""
Base Exception thrown by the Twitter object when there is a
"""
pass
+
class TwitterHTTPError(TwitterError):
"""
Exception thrown by the Twitter object when there is an
fmt = ("." + self.format) if self.format else ""
return (
"Twitter sent status %i for URL: %s%s using parameters: "
- "(%s)\ndetails: %s" %(
+ "(%s)\ndetails: %s" % (
self.e.code, self.uri, fmt, self.uriparts,
self.response_data))
+
class TwitterResponse(object):
"""
Response from a twitter request. Behaves like a list or a string
def __init__(self, response, headers):
response_typ.__init__(self, response)
TwitterResponse.__init__(self, headers)
+
def __new__(cls, response, headers):
return response_typ.__new__(cls, response)
-
return WrappedTwitterResponse(response, headers)
-
class TwitterCall(object):
def __init__(
- self, auth, format, domain, callable_cls, uri="",
- uriparts=None, secure=True):
+ self, auth, format, domain, callable_cls, uri="",
+ uriparts=None, secure=True):
self.auth = auth
self.format = format
self.domain = domain
def extend_call(arg):
return self.callable_cls(
auth=self.auth, format=self.format, domain=self.domain,
- callable_cls=self.callable_cls, uriparts=self.uriparts \
- + (arg,),
+ callable_cls=self.callable_cls, uriparts=self.uriparts
+ + (arg,),
secure=self.secure)
if k == "_":
return extend_call
# the list of uriparts, assume the id goes at the end.
id = kwargs.pop('id', None)
if id:
- uri += "/%s" %(id)
+ uri += "/%s" % (id)
# If an _id kwarg is present, this is treated as id as a CGI
# param.
dot = ""
if self.format:
dot = "."
- uriBase = "http%s://%s/%s%s%s" %(
- secure_str, self.domain, uri, dot, self.format)
+ uriBase = "http%s://%s/%s%s%s" % (
+ secure_str, self.domain, uri, dot, self.format)
headers = {'Accept-Encoding': 'gzip'}
if self.auth:
handle = urllib_request.urlopen(req, **kwargs)
if handle.headers['Content-Type'] in ['image/jpeg', 'image/png']:
return handle
- elif handle.info().get('Content-Encoding') == 'gzip':
+ try:
+ data = handle.read()
+ except httplib.IncompleteRead as e:
+ # Even if we don't get all the bytes we should have there
+ # may be a complete response in e.partial
+ data = e.partial
+ if handle.info().get('Content-Encoding') == 'gzip':
# Handle gzip decompression
- buf = StringIO(handle.read())
+ buf = StringIO(data)
f = gzip.GzipFile(fileobj=buf)
data = f.read()
- else:
- data = handle.read()
-
if "json" == self.format:
res = json.loads(data.decode('utf8'))
return wrap_response(res, handle.headers)
else:
raise TwitterHTTPError(e, uri, self.format, arg_data)
+
class Twitter(TwitterCall):
"""
The minimalist yet fully featured Twitter API class.
t = Twitter(
auth=OAuth(token, token_key, con_secret, con_secret_key)))
- # Get the public timeline
- t.statuses.public_timeline()
+ # Get your "home" timeline
+ t.statuses.home_timeline()
# Get a particular friend's timeline
t.statuses.friends_timeline(id="billybob")
Searching Twitter::
- twitter_search = Twitter(domain="search.twitter.com")
-
- # Find the latest search trends
- twitter_search.trends()
-
- # Search for the latest News on #gaza
- twitter_search.search(q="#gaza")
+ # Search for the latest tweets about #pycon
+ t.search.tweets(q="#pycon")
Using the data returned
Twitter API calls return decoded JSON. This is converted into
a bunch of Python lists, dicts, ints, and strings. For example::
- x = twitter.statuses.public_timeline()
+ x = twitter.statuses.home_timeline()
# The first 'tweet' in the timeline
x[0]
"""
def __init__(
- self, format="json",
- domain="api.twitter.com", secure=True, auth=None,
- api_version=_DEFAULT):
+ self, format="json",
+ domain="api.twitter.com", secure=True, auth=None,
+ api_version=_DEFAULT):
"""
Create a new twitter API connector.
`domain` lets you change the domain you are connecting. By
- default it's `api.twitter.com` but `search.twitter.com` may be
- useful too.
+ default it's `api.twitter.com`.
If `secure` is False you will connect with HTTP instead of
HTTPS.
`api_version` is used to set the base uri. By default it's
- '1'. If you are using "search.twitter.com" set this to None.
+ '1.1'.
"""
if not auth:
auth = NoAuth()
if (format not in ("json", "xml", "")):
- raise ValueError("Unknown data format '%s'" %(format))
+ raise ValueError("Unknown data format '%s'" % (format))
if api_version is _DEFAULT:
if domain == 'api.twitter.com':