code/twython
1
Fork 0
Code Issues 23 Pull requests 4 Projects Releases 5 Packages Wiki Activity Actions

Done with implementation of twython-requests, unit testing remaining

Browse source
This commit is contained in:
kracekumar 2011-12-06 00:12:17 +05:30
parent d3c364601b
commit 7334556248
3 changed files with 347 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

17
twython/tests.py Normal file
View file

@ -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()

6
twython/twython.py
View file

@ -210,7 +210,7 @@ class Twython(object):
fn = api_table[api_call]
base = re.sub(
'\{\{(?P<m>[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.

342
twython/twythonrequests.py
View file

@ -10,8 +10,8 @@
Questions, comments? ryan@venodesigns.net, me@kracekumar.com
"""
__author__ = "kracekumar"
__version__ = "0.1"
__author__ = "kracekumar <me@kracekumar.com>"
__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)