twitter/api.py

   1 try:
   2     import urllib.request as urllib_request
   3     import urllib.error as urllib_error
   4 except ImportError:
   5     import urllib2 as urllib_request
   6     import urllib2 as urllib_error
   7
   8 try:
   9     from cStringIO import StringIO
  10 except ImportError:
  11     from io import BytesIO as StringIO
  12
  13 from .twitter_globals import POST_ACTIONS
  14 from .auth import NoAuth
  15
  16 import re
  17 import gzip
  18
  19 try:
  20     import http.client as http_client
  21 except ImportError:
  22     import httplib as http_client
  23
  24 import json
  25
  26
  27 class _DEFAULT(object):
  28     pass
  29
  30 class TwitterError(Exception):
  31     """
  32     Base Exception thrown by the Twitter object when there is a
  33     general error interacting with the API.
  34     """
  35     pass
  36
  37 class TwitterHTTPError(TwitterError):
  38     """
  39     Exception thrown by the Twitter object when there is an
  40     HTTP error interacting with twitter.com.
  41     """
  42     def __init__(self, e, uri, format, uriparts):
  43         self.e = e
  44         self.uri = uri
  45         self.format = format
  46         self.uriparts = uriparts
  47         try:
  48             data = self.e.fp.read()
  49         except http_client.IncompleteRead as e:
  50             # can't read the error text
  51             # let's try some of it
  52             data = e.partial
  53         if self.e.headers.get('Content-Encoding') == 'gzip':
  54             buf = StringIO(data)
  55             f = gzip.GzipFile(fileobj=buf)
  56             self.response_data = f.read()
  57         else:
  58             self.response_data = data
  59
  60     def __str__(self):
  61         fmt = ("." + self.format) if self.format else ""
  62         return (
  63             "Twitter sent status %i for URL: %s%s using parameters: "
  64             "(%s)\ndetails: %s" %(
  65                 self.e.code, self.uri, fmt, self.uriparts,
  66                 self.response_data))
  67
  68 class TwitterResponse(object):
  69     """
  70     Response from a twitter request. Behaves like a list or a string
  71     (depending on requested format) but it has a few other interesting
  72     attributes.
  73
  74     `headers` gives you access to the response headers as an
  75     httplib.HTTPHeaders instance. You can do
  76     `response.headers.get('h')` to retrieve a header.
  77     """
  78     def __init__(self, headers):
  79         self.headers = headers
  80
  81     @property
  82     def rate_limit_remaining(self):
  83         """
  84         Remaining requests in the current rate-limit.
  85         """
  86         return int(self.headers.get('X-Rate-Limit-Remaining', "0"))
  87
  88     @property
  89     def rate_limit_limit(self):
  90         """
  91         The rate limit ceiling for that given request.
  92         """
  93         return int(self.headers.get('X-Rate-Limit-Limit', "0"))
  94
  95     @property
  96     def rate_limit_reset(self):
  97         """
  98         Time in UTC epoch seconds when the rate limit will reset.
  99         """
 100         return int(self.headers.get('X-Rate-Limit-Reset', "0"))
 101
 102
 103 def wrap_response(response, headers):
 104     response_typ = type(response)
 105     if response_typ is bool:
 106         # HURF DURF MY NAME IS PYTHON AND I CAN'T SUBCLASS bool.
 107         response_typ = int
 108     elif response_typ is str:
 109         return response
 110
 111     class WrappedTwitterResponse(response_typ, TwitterResponse):
 112         __doc__ = TwitterResponse.__doc__
 113
 114         def __init__(self, response, headers):
 115             response_typ.__init__(self, response)
 116             TwitterResponse.__init__(self, headers)
 117         def __new__(cls, response, headers):
 118             return response_typ.__new__(cls, response)
 119
 120     return WrappedTwitterResponse(response, headers)
 121
 122
 123
 124 class TwitterCall(object):
 125
 126     def __init__(
 127         self, auth, format, domain, callable_cls, uri="",
 128         uriparts=None, secure=True, timeout=None):
 129         self.auth = auth
 130         self.format = format
 131         self.domain = domain
 132         self.callable_cls = callable_cls
 133         self.uri = uri
 134         self.uriparts = uriparts
 135         self.secure = secure
 136         self.timeout = timeout
 137
 138     def __getattr__(self, k):
 139         try:
 140             return object.__getattr__(self, k)
 141         except AttributeError:
 142             def extend_call(arg):
 143                 return self.callable_cls(
 144                     auth=self.auth, format=self.format, domain=self.domain,
 145                     callable_cls=self.callable_cls, timeout=self.timeout, uriparts=self.uriparts \
 146                         + (arg,),
 147                     secure=self.secure)
 148             if k == "_":
 149                 return extend_call
 150             else:
 151                 return extend_call(k)
 152
 153     def __call__(self, **kwargs):
 154         # Build the uri.
 155         uriparts = []
 156         for uripart in self.uriparts:
 157             # If this part matches a keyword argument, use the
 158             # supplied value otherwise, just use the part.
 159             uriparts.append(str(kwargs.pop(uripart, uripart)))
 160         uri = '/'.join(uriparts)
 161
 162         method = kwargs.pop('_method', None)
 163         if not method:
 164             method = "GET"
 165             for action in POST_ACTIONS:
 166                 if re.search("%s(/\d+)?$" % action, uri):
 167                     method = "POST"
 168                     break
 169
 170         # If an id kwarg is present and there is no id to fill in in
 171         # the list of uriparts, assume the id goes at the end.
 172         id = kwargs.pop('id', None)
 173         if id:
 174             uri += "/%s" %(id)
 175
 176         # If an _id kwarg is present, this is treated as id as a CGI
 177         # param.
 178         _id = kwargs.pop('_id', None)
 179         if _id:
 180             kwargs['id'] = _id
 181
 182         # If an _timeout is specified in kwargs, use it
 183         _timeout = kwargs.pop('_timeout', None)
 184
 185         secure_str = ''
 186         if self.secure:
 187             secure_str = 's'
 188         dot = ""
 189         if self.format:
 190             dot = "."
 191         uriBase = "http%s://%s/%s%s%s" %(
 192                     secure_str, self.domain, uri, dot, self.format)
 193
 194         headers = {'Accept-Encoding': 'gzip'}
 195         if self.auth:
 196             headers.update(self.auth.generate_headers())
 197             arg_data = self.auth.encode_params(uriBase, method, kwargs)
 198             if method == 'GET':
 199                 uriBase += '?' + arg_data
 200                 body = None
 201             else:
 202                 body = arg_data.encode('utf8')
 203
 204         req = urllib_request.Request(uriBase, body, headers)
 205         return self._handle_response(req, uri, arg_data, _timeout)
 206
 207     def _handle_response(self, req, uri, arg_data, _timeout=None):
 208         kwargs = {}
 209         if _timeout:
 210             kwargs['timeout'] = _timeout
 211         try:
 212             handle = urllib_request.urlopen(req, **kwargs)
 213             if handle.headers['Content-Type'] in ['image/jpeg', 'image/png']:
 214                 return handle
 215             try:
 216                 data = handle.read()
 217             except http_client.IncompleteRead as e:
 218                 # Even if we don't get all the bytes we should have there
 219                 # may be a complete response in e.partial
 220                 data = e.partial
 221             if handle.info().get('Content-Encoding') == 'gzip':
 222                 # Handle gzip decompression
 223                 buf = StringIO(data)
 224                 f = gzip.GzipFile(fileobj=buf)
 225                 data = f.read()
 226             if "json" == self.format:
 227                 res = json.loads(data.decode('utf8'))
 228                 return wrap_response(res, handle.headers)
 229             else:
 230                 return wrap_response(
 231                     data.decode('utf8'), handle.headers)
 232         except urllib_error.HTTPError as e:
 233             if (e.code == 304):
 234                 return []
 235             else:
 236                 raise TwitterHTTPError(e, uri, self.format, arg_data)
 237
 238 class Twitter(TwitterCall):
 239     """
 240     The minimalist yet fully featured Twitter API class.
 241
 242     Get RESTful data by accessing members of this class. The result
 243     is decoded python objects (lists and dicts).
 244
 245     The Twitter API is documented at:
 246
 247       http://dev.twitter.com/doc
 248
 249
 250     Examples::
 251
 252         t = Twitter(
 253             auth=OAuth(token, token_key, con_secret, con_secret_key)))
 254
 255         # Get your "home" timeline
 256         t.statuses.home_timeline()
 257
 258         # Get a particular friend's timeline
 259         t.statuses.friends_timeline(id="billybob")
 260
 261         # Also supported (but totally weird)
 262         t.statuses.friends_timeline.billybob()
 263
 264         # Update your status
 265         t.statuses.update(
 266             status="Using @sixohsix's sweet Python Twitter Tools.")
 267
 268         # Send a direct message
 269         t.direct_messages.new(
 270             user="billybob",
 271             text="I think yer swell!")
 272
 273         # Get the members of tamtar's list "Things That Are Rad"
 274         t._("tamtar")._("things-that-are-rad").members()
 275
 276         # Note how the magic `_` method can be used to insert data
 277         # into the middle of a call. You can also use replacement:
 278         t.user.list.members(user="tamtar", list="things-that-are-rad")
 279
 280         # An *optional* `_timeout` parameter can also be used for API
 281         # calls which take much more time than normal or twitter stops
 282         # responding for some reasone
 283         t.users.lookup(
 284             screen_name=','.join(A_LIST_OF_100_SCREEN_NAMES), \
 285             _timeout=1)
 286
 287
 288
 289     Searching Twitter::
 290
 291         # Search for the latest tweets about #pycon
 292         t.search.tweets(q="#pycon")
 293
 294
 295     Using the data returned
 296     -----------------------
 297
 298     Twitter API calls return decoded JSON. This is converted into
 299     a bunch of Python lists, dicts, ints, and strings. For example::
 300
 301         x = twitter.statuses.home_timeline()
 302
 303         # The first 'tweet' in the timeline
 304         x[0]
 305
 306         # The screen name of the user who wrote the first 'tweet'
 307         x[0]['user']['screen_name']
 308
 309
 310     Getting raw XML data
 311     --------------------
 312
 313     If you prefer to get your Twitter data in XML format, pass
 314     format="xml" to the Twitter object when you instantiate it::
 315
 316         twitter = Twitter(format="xml")
 317
 318     The output will not be parsed in any way. It will be a raw string
 319     of XML.
 320
 321     """
 322     def __init__(
 323         self, format="json",
 324         domain="api.twitter.com", secure=True, auth=None,
 325         api_version=_DEFAULT):
 326         """
 327         Create a new twitter API connector.
 328
 329         Pass an `auth` parameter to use the credentials of a specific
 330         user. Generally you'll want to pass an `OAuth`
 331         instance::
 332
 333             twitter = Twitter(auth=OAuth(
 334                     token, token_secret, consumer_key, consumer_secret))
 335
 336
 337         `domain` lets you change the domain you are connecting. By
 338         default it's `api.twitter.com` but `search.twitter.com` may be
 339         useful too.
 340
 341         If `secure` is False you will connect with HTTP instead of
 342         HTTPS.
 343
 344         `api_version` is used to set the base uri. By default it's
 345         '1'. If you are using "search.twitter.com" set this to None.
 346         """
 347         if not auth:
 348             auth = NoAuth()
 349
 350         if (format not in ("json", "xml", "")):
 351             raise ValueError("Unknown data format '%s'" %(format))
 352
 353         if api_version is _DEFAULT:
 354             api_version = '1.1'
 355
 356         uriparts = ()
 357         if api_version:
 358             uriparts += (str(api_version),)
 359
 360         TwitterCall.__init__(
 361             self, auth=auth, format=format, domain=domain,
 362             callable_cls=TwitterCall,
 363             secure=secure, uriparts=uriparts)
 364
 365
 366 __all__ = ["Twitter", "TwitterError", "TwitterHTTPError", "TwitterResponse"]