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