X-Git-Url: https://jfr.im/git/z_archive/twitter.git/blobdiff_plain/9b0c5df39edd3adfa7f38885b4fa3f4b0ccbef53..7bfe7d97d94ea248b1d73c746354cd241203fe5d:/twitter/api.py diff --git a/twitter/api.py b/twitter/api.py index 42d9e53..f675822 100644 --- a/twitter/api.py +++ b/twitter/api.py @@ -1,10 +1,7 @@ - -import urllib2 - -from exceptions import Exception +import urllib.request, urllib.error, urllib.parse from twitter.twitter_globals import POST_ACTIONS -from twitter.auth import UserPassAuth, NoAuth +from twitter.auth import NoAuth def _py26OrGreater(): import sys @@ -27,26 +24,75 @@ class TwitterHTTPError(TwitterError): Exception thrown by the Twitter object when there is an HTTP error interacting with twitter.com. """ - def __init__(self, e, uri, format, encoded_args): + def __init__(self, e, uri, format, uriparts): self.e = e self.uri = uri self.format = format - self.encoded_args = encoded_args + self.uriparts = uriparts def __str__(self): - return "Twitter sent status %i for URL: %s.%s using parameters: (%s)\ndetails: %s" %( - self.e.code, self.uri, self.format, self.encoded_args, self.e.fp.read()) + return ( + "Twitter sent status %i for URL: %s.%s using parameters: " + "(%s)\ndetails: %s" %( + self.e.code, self.uri, self.format, self.uriparts, + self.e.fp.read())) + +class TwitterResponse(object): + """ + 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. + """ + def __init__(self, headers): + self.headers = headers + + @property + def rate_limit_remaining(self): + """ + Remaining requests in the current rate-limit. + """ + return int(self.headers.getheader('X-RateLimit-Remaining')) + + @property + def rate_limit_reset(self): + """ + Time in UTC epoch seconds when the rate limit will reset. + """ + return int(self.headers.getheader('X-RateLimit-Reset')) + + +def wrap_response(response, headers): + response_typ = type(response) + if response_typ is bool: + # HURF DURF MY NAME IS PYTHON AND I CAN'T SUBCLASS bool. + response_typ = int + + class WrappedTwitterResponse(response_typ, TwitterResponse): + __doc__ = TwitterResponse.__doc__ + + def __init__(self, response): + if response_typ is not int: + response_typ.__init__(self, response) + TwitterResponse.__init__(self, headers) + + return WrappedTwitterResponse(response) + + class TwitterCall(object): def __init__( self, auth, format, domain, uri="", agent=None, - encoded_args=None, secure=True): + uriparts=None, secure=True): self.auth = auth self.format = format self.domain = domain self.uri = uri self.agent = agent - self.encoded_args = encoded_args + self.uriparts = uriparts self.secure = secure def __getattr__(self, k): @@ -54,61 +100,65 @@ class TwitterCall(object): return object.__getattr__(self, k) except AttributeError: return TwitterCall( - self.auth, self.format, self.domain, - self.uri + "/" + k, self.agent, self.encoded_args, self.secure) + auth=self.auth, format=self.format, domain=self.domain, + agent=self.agent, uriparts=self.uriparts + (k,), + secure=self.secure) def __call__(self, **kwargs): - uri = self.uri.strip("/") + # Build the uri. + uriparts = [] + for uripart in self.uriparts: + # If this part matches a keyword argument, use the + # supplied value otherwise, just use the part. + uriparts.append(str(kwargs.pop(uripart, uripart))) + uri = '/'.join(uriparts) + method = "GET" for action in POST_ACTIONS: - if self.uri.endswith(action): + if uri.endswith(action): method = "POST" - if (self.agent): - kwargs["source"] = self.agent break + # 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. + id = kwargs.pop('id', None) + if id: + uri += "/%s" %(id) + secure_str = '' if self.secure: secure_str = 's' dot = "" - if self.format != '': + if self.format: dot = "." uriBase = "http%s://%s/%s%s%s" %( secure_str, self.domain, uri, dot, self.format) - if (not self.encoded_args): - if kwargs.has_key('id'): - uri += "/%s" %(kwargs['id']) - - self.encoded_args = self.auth.encode_params(uriBase, method, kwargs) - - argStr = "" - argData = None - if (method == "GET"): - if self.encoded_args: - argStr = "?%s" %(self.encoded_args) - else: - argData = self.encoded_args - headers = {} - if (self.agent): - headers["X-Twitter-Client"] = self.agent - if self.auth is not None: + if self.auth: headers.update(self.auth.generate_headers()) + arg_data = self.auth.encode_params(uriBase, method, kwargs) + if method == 'GET': + uriBase += '?' + arg_data + body = None + else: + body = arg_data.encode('utf8') + + req = urllib.request.Request(uriBase, body, headers) - req = urllib2.Request(uriBase+argStr, argData, headers) - try: - handle = urllib2.urlopen(req) + handle = urllib.request.urlopen(req) if "json" == self.format: - return json.loads(handle.read()) + res = json.loads(handle.read().decode('utf8')) + return wrap_response(res, handle.headers) else: - return handle.read() - except urllib2.HTTPError, e: + return wrap_response( + handle.read().decode('utf8'), handle.headers) + except urllib.error.HTTPError as e: if (e.code == 304): return [] else: - raise TwitterHTTPError(e, uri, self.format, self.encoded_args) + raise TwitterHTTPError(e, uri, self.format, arg_data) class Twitter(TwitterCall): """ @@ -119,12 +169,13 @@ class Twitter(TwitterCall): The Twitter API is documented here: - http://apiwiki.twitter.com/ - http://groups.google.com/group/twitter-development-talk/web/api-documentation + http://dev.twitter.com/doc + Examples:: - twitter = Twitter("hello@foo.com", "password123") + twitter = Twitter( + auth=OAuth(token, token_key, con_secret, con_secret_key))) # Get the public timeline twitter.statuses.public_timeline() @@ -140,6 +191,10 @@ class Twitter(TwitterCall): user="billybob", text="I think yer swell!") + # Get the members of a particular list of a particular friend + twitter.user.listname.members(user="billybob", listname="billysbuds") + + Searching Twitter:: twitter_search = Twitter(domain="search.twitter.com") @@ -150,10 +205,12 @@ class Twitter(TwitterCall): # Search for the latest News on #gaza twitter_search.search(q="#gaza") - 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, + 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() @@ -163,37 +220,63 @@ class Twitter(TwitterCall): # The screen name of the user who wrote the first 'tweet' x[0]['user']['screen_name'] - Getting raw XML data:: - If you prefer to get your Twitter data in XML format, pass - format="xml" to the Twitter object when you instantiate it: + Getting raw XML data + -------------------- + + If you prefer to get your Twitter data in XML format, pass + format="xml" to the Twitter object when you instantiate it:: twitter = Twitter(format="xml") The output will not be parsed in any way. It will be a raw string of XML. + """ def __init__( - self, email=None, password=None, format="json", domain="twitter.com", - agent=None, secure=True, auth=None): - """ - Create a new twitter API connector using the specified - credentials (email and password). Format specifies the output - format ("json" (default) or "xml"). + self, format="json", + domain="twitter.com", secure=True, auth=None, + api_version=''): """ - - if email is not None or password is not None: - if auth is not None: - raise ValueError, "can't specify 'email' or 'password' and 'auth' params" - auth = UserPassAuth(email, password) + Create a new twitter API connector. + Pass an `auth` parameter to use the credentials of a specific + user. Generally you'll want to pass an `OAuth` + instance:: + + twitter = Twitter(auth=OAuth( + token, token_secret, consumer_key, consumer_secret)) + + + `domain` lets you change the domain you are connecting. By + default it's twitter.com but `search.twitter.com` may be + useful too. + + If `secure` is False you will connect with HTTP instead of + HTTPS. + + The value of `agent` is sent in the `X-Twitter-Client` + header. This is deprecated. Instead Twitter determines the + application using the OAuth Client Key and Client Key Secret + parameters. + + `api_version` is used to set the base uri. By default it's + nothing, but if you set it to '1' your URI will start with + '1/'. + """ if not auth: auth = NoAuth() if (format not in ("json", "xml", "")): - raise TwitterError("Unknown data format '%s'" %(format)) + raise ValueError("Unknown data format '%s'" %(format)) + + uriparts = () + if api_version: + uriparts += (str(api_version),) + TwitterCall.__init__( - self, auth, format, domain, "", agent, - secure=secure) + self, auth=auth, format=format, domain=domain, + secure=secure, uriparts=uriparts) + -__all__ = ["Twitter", "TwitterError", "TwitterHTTPError"] +__all__ = ["Twitter", "TwitterError", "TwitterHTTPError", "TwitterResponse"]