From 733455624875b2492a657be4d57d18103adb1b1c Mon Sep 17 00:00:00 2001 From: kracekumar Date: Tue, 6 Dec 2011 00:12:17 +0530 Subject: [PATCH] Done with implementation of twython-requests, unit testing remaining --- twython/tests.py | 17 ++ twython/twython.py | 6 +- twython/twythonrequests.py | 342 +++++++++++++++++++++++++++++++++++-- 3 files changed, 347 insertions(+), 18 deletions(-) create mode 100644 twython/tests.py diff --git a/twython/tests.py b/twython/tests.py new file mode 100644 index 0000000..dc3b766 --- /dev/null +++ b/twython/tests.py @@ -0,0 +1,17 @@ +#! /usr/bin/env python +# -*- coding: utf-8 -*- +import unittest +import json +import requests + +import twythonrequests + +class TestTwythonRequests(unittest.TestCase): + def test_search(self): + twitter = twythonrequests.Twython() + result = twitter.search(q='python') + self.assertTrue(result['results']) + + +if __name__ == "__main__": + unittest.main() diff --git a/twython/twython.py b/twython/twython.py index 2c643be..4241bad 100644 --- a/twython/twython.py +++ b/twython/twython.py @@ -210,7 +210,7 @@ class Twython(object): fn = api_table[api_call] base = re.sub( '\{\{(?P[a-zA-Z_]+)\}\}', - lambda m: "%s" % kwargs.get(m.group(1), '1'),\ + lambda m: "%s" % kwargs.get(m.group(1), '1'),\ # The '1' here catches the API version. Slightly hilarious. base_url + fn['url'] ) @@ -460,7 +460,7 @@ class Twython(object): e.code, e.code) def isListSubscriber(self, username, list_id, id, version = 1): - """ isListSubscriber(self, list_id, id, version) + """ isListSubscriber(self, username, list_id, id, version) Check if a specified user (id) is a subscriber of the list in \ question (list_id). @@ -531,7 +531,7 @@ class Twython(object): Updates the authenticating user's profile image (avatar). Parameters: - image - Required. Must be a valid GIF, JPG, or PNG image of \ + filename - Required. Must be a valid GIF, JPG, or PNG image of \ less than 700 kilobytes in size. Images with width larger than \ 500 pixels will be scaled down. diff --git a/twython/twythonrequests.py b/twython/twythonrequests.py index 4f3cccb..bd0bf03 100644 --- a/twython/twythonrequests.py +++ b/twython/twythonrequests.py @@ -10,8 +10,8 @@ Questions, comments? ryan@venodesigns.net, me@kracekumar.com """ -__author__ = "kracekumar" -__version__ = "0.1" +__author__ = "kracekumar " +__version__ = "0.0.1" """ Importing requests and requests-oauth. @@ -23,7 +23,7 @@ import mimetools import re import inspect from urlparse import parse_qs - +from urllib2 import HTTPError try: #Python 2.6 and up import requests @@ -48,7 +48,7 @@ except ImportError: #imports required for Error Handling. Now I am importing from twython file in # future, it will be include in the same file, Hope ryan agrees :) -from twython import TwythonError, APILimit, RateLimitError, AuthError +from twython import TwythonError, APILimit, AuthError class Twython(object): def __init__(self, twitter_token = None, twitter_secret = None, \ @@ -127,6 +127,7 @@ class Twython(object): pass def get_authorized_tokens(self): + return self.get_autheentication_tokens(internal = 1) def get_authentication_tokens(self, internal = None): """ @@ -144,7 +145,7 @@ class Twython(object): self.response = parse_qs(self.response.content) try: return {'oauth_token': self.response['oauth_token'],\ - 'oauth_secret': self.response['oauth_token_secret'])} + 'oauth_secret': self.response['oauth_token_secret']} except AttributeError, e: raise TwythonError("Something went wrong, with parsing or call\ \n get_authentication_token() failed with %s error code"% \ @@ -154,15 +155,326 @@ class Twython(object): Error Message:%s"%(self.response.status_code,\ self.response.error.msg)) + @staticmethod + def constructApiURL(base_url, params): + """ + We don't need this since requests will build url for us + """ + return base_url + "?" + "&".join(["%s=%s" %(Twython.unicode2utf8(key),\ + urllib.quote_plus(Twython.unicode2utf8(value)))\ + for (key, value) in params.iteritems()]) + + @staticmethod + def shortenURL(url_to_shorten, shortener = "http://is.gd/api.php",\ + query = "longurl"): + """ + Nowadays twitter automatically shortens urls in the tweets, so + there isn't real need to shorten url, unless explicitly you want to + use your own fav services like goo.gl. + + We will implement this one soon + """ + pass + + def bulkUserLookup(self, ids = None, screen_names = None, version = 1,\ + **kwargs): + """ + bulkUserLookup(self, ids = None, screen_names = None, version = 1, + **kwargs): + + A method to do bulk user lookups against the Twitter API. + Arguments (ids (numbers) / screen_names (strings)) should be flat. + Arrays that contain their respective data sets. + + Statuses for the users in question will be returned inline if they + exists. Requires authentication! + """ + if ids: + kwargs['user_id'] = ','.join(map(str, ids)) + if screen_names: + kwargs['screen_name'] = ','.join(screen_names) + + try: + self.response = requests.post("\ + http://api.twitter.com/%d/users/lookup.json" %version,\ + kwargs) + self.response.raise_for_status() + return json.loads(self.response.content) + except HTTPError, e: + raise TwythonError("bulkUserLookup() failed with a %s error code."\ + % `e.code`, e.code) + + def search(self, **kwargs): + """ + search(**kwargs) + + + Returns Tweets that match the specified query. + + Parameters: + + q: query to search for example + + See the documentation at http://dev.twitter.com/doc/get/ + search. + + Pass in the API supported arguments as named parameters. + + e.g: x.search(q='python', page='2') + + """ + try: + self.response = requests.post(\ + "http://search.twitter.com/search.json",\ + data=kwargs) + self.response.raise_for_status() + return json.loads(self.response.content) + except HTTPError, e: + raise TwythonError("search() failed with %s error code" \ + % `e.code`, e.code) + + def searchTwitter(self, **kwargs): + """ + use search(). This will be removed soon. + """ + return self.search(**kwargs) + + def searchGen(self, **kwargs): + """ + seaarchGen(self, **kwargs) + + Returns a generator of tweets that match a specified query. + + Documentation: http://dev.twitter.com/doc/get/search + + e.g: x.searchGen(q='python', page='2') + + """ + try: + self.response = self.search(kwargs) + self.response.raise_for_status() + self.response = json.loads(self.response.content) + except HTTPError, e: + raise TwythonError("searchGen() exited with %d status code and \ + code "%self.response.status_code, e.code) + + if self.response['results']: + raise StopIteration + else: + for tweet in data['results']: + yield tweet + + if 'page' not in kwargs: + kwargs['page'] = 2 + else: + try: + # This line converts page param in query parameter to int and + # adds one because we are running inside func which will yield + # list of tweet using generator and converts to string. + kwargs['page'] = str(int(kwargs['page']) + 1) + except TypeError: + raise TwythonError("searchGen() exited because it page \ + takes string ") + except e: + raise TwythonError("searchGen() failed with %s error code"%\ + `e.code`, e.code) + + for tweet in self.searchGen(**kwargs): + yield tweet + + + def searchTwitterGen(self, **kwargs): + """ + use searchGen(). This will be removed soon. + """ + return self.searchGen(kwargs) + + def isListMember(self, list_id, id, username, version = 1): + """ + isListMember(self, list_id, id, username, version =1) + + Check if a specified user(id) is a member of the list in question + (list_id) + + **Note: This method may not work for private/protected lists, + unless you're authenticated and have access to those lists. + + Parameters: + list_id - Required. The slug of the list to check against. + id - Required. The ID of the user being checked in the list. + username - User who owns the list you're checking against\ + (username) version(number) - Optional. API version to \ + request.\ + + version (number) - Optional. API version to request. + + Entire Twython class defaults to 1, but you can override on\ + + a function-by-function or class basis - (version=2), etc. + """ + try: + self.response = requests.post("http://api.twitter.com/%d/%s/%s/\ + members/%s.json" % (version, username, list_id,\ + `id`), headers = self.headers) + self.response.raise_for_status() + return json.loads(Self.response.content) + except HTTPError, e: + raise TwythonError("isListMember() failed with status code %d"\ + %self.response.status_code, e.code) + + def isListSubscriber(self, username, list_id, version = 1): + """ + isListSubscriber(self, username, list_id, id, version) + + Check if a specified user(id) is a subscriber of the list in \ + question(list_id) + + **Note: This method may not work for private/protected lists, + unless you're authenticated and have access to those lists. + + Parameters: + list_id - Required. The slug of the list to check against. + id - Required. The ID of the user being checked in the list. + username - Required. The username of the owner of the list\ + that you're seeing if someone is subscribed to. + + version (number) - Optional. API version to request. + Entire Twython class defaults to 1, but you can override on\ + a function-by-function or class basis - (version=2), etc. + """ + try: + self.response = requests.post("http://api.twitter.com/%d/%s/%s\ + following/%s.json" % (version,username,list_id,\ + `id`), headers = self.headers) + self.response.raise_for_status() + return json.loads(self.response.content) + except HTTPError, e: + raise TwythonError("isListMember() failed with %d error code."%\ + self.response.status_code, e.code) + + def updateProfileBackgroundImage(self,filename,tile="true",version=1): + """ + updateProfileBackgroundImage(filename,tile="true") + + Updates the authenticating user's profile background image. + + Parameters: + image - Required. Must be a valid GIF, JPG, or PNG image of\ + less than 800 kilobytes in size. Images with with larger \ + than 2048 pixels will be forceably scaled down. + + tile - Optional (defaults to true). If set to true the \ + background image will be displayed tiled. + The image will not be tiled otherwise. + + ** Note: It's sad, but when using this method, pass the \ + tile value as string, + + e.g. title="false" + + version (number) - Optional. API version to request.\ + Entire Twython class defaults to 1, but you can override on\ + a function-by-function or class basis - (version=2), etc. + + """ + url = "http://api.twitter.com/%d/account/update_profile_background.\ + json?tile=%s" %(version,tile) + try: + files = {filename: open(filename, 'rb')} + except IOError, e: + raise TwythonError("file reading %d error"%`e.code`, e.code) + + try: + self.response = request.post(url, files=files) + self.response.raise_for_status() + return self.response.status_code + except HTTPError, e: + raise Twython("updateProfileBackgroundImage failed with %d\ + error code"%self.response.status_code, e.code) + + def updateProfileImage(self,filename,version=1): + """ + updateProfileImage(filename) + + Updates the authenticating user's profile image (avatar). + + Parameters: + image - Required. Must be valid GIF, JPG, or PNG image of\ + less than 700 kilobytes in size. Image with width larger \ + than 500 pixels will be scaled down. + + version (number) - Optional. API version to request. + + Entire Twython class defaults to 1, but you can override on\ + a function-by-function or class basis - (version=2), etc. + + """ + url = "http://api.twitter.com/%d/account/update_profile_image.json"\ + %version + try: + files = {filename: open(filename, 'rb')} + except IOError, e: + raise TwythonError("file reading %d error"%`e.code`, e.code) + + try: + self.response = requests.post(url, files=files) + self.response.raise_for_status() + return self.response.status_code + except HTTPError, e: + raise TwythonError("updateProfileImage() failed with %d error\ + code"% self.response.status_code, e.code) + + def getProfileImageUrl(self, username, size=None, version=1): + """ + getProfileImageUrl(username) + + Gets the URL for the user's profile image. + + Parameters: + username - Required. User name of the user you want the image + url of. + + size - Optional.Options 'normal', 'mini', 'bigger'. Defaults + to 'normal' if not given. + + version (number) - Optional. API version to request. + + Entire Twython class defaults to 1, but you can override on\ + a function-by-function or class basis - (version=2), etc. + + """ + url = "http://api.twitter.com/%s/users/profile_image/%s.json"%\ + (version, username) + try: + self.response = requests.post(url, size=size) + self.raise_for_status() + if self.response.status_code in (301, 302, 303, 307): + return self.response['location'] + else: + return json.loads(self.response.content) + except: + return TwythonError("getProfileIMageUrl() failed with %d \ + error code"% self.response.status_code) + + @staticmethod + def unicode2utf8(text): + try: + if isinstance(text, unicode): + text = text.encode('utf-8') + except: + pass + return text + + @staticmethod + def encode(text): + if isinstance(text, (str, unicode)): + return Twython.unicode2utf8(text) + return str(text) + + + + + - - - - - - - - - - +