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

Aiming to remove need for httplib2, but sadly this commit is broken.

Browse source
This commit is contained in:
Ryan McGrath 2009-05-26 01:29:03 -04:00
parent 10c32e427c
commit 1887280855
1 changed files with 100 additions and 105 deletions
Download patch file Download diff file Expand all files Collapse all files

187
tango.py
View file

@ -7,114 +7,109 @@
import urllib, urllib2 import urllib, urllib2
try: try:
import simplejson import simplejson
except: except:
print "Tango requires the simplejson library to work. http://www.undefined.org/python/" print "Tango requires the simplejson library to work. http://www.undefined.org/python/"
# Should really deprecate httplib2 at some point...
try:
import httplib2
except:
print "Tango requires httplib2 for authentication purposes. http://code.google.com/p/httplib2/"
try: try:
import oauth import oauth
except: except:
print "Tango requires oauth for authentication purposes. http://oauth.googlecode.com/svn/code/python/oauth/oauth.py" print "Tango requires oauth for authentication purposes. http://oauth.googlecode.com/svn/code/python/oauth/oauth.py"
# Need to support URL shortening
class setup: class setup:
def __init__(self, authtype = "OAuth", username = None, password = None, oauth_keys = None): def __init__(self, authtype = "OAuth", username = None, password = None, oauth_keys = None):
self.authtype = authtype self.authtype = authtype
self.authenticated = False self.authenticated = False
self.username = username self.username = username
self.password = password self.password = password
self.oauth_keys = oauth_keys self.oauth_keys = oauth_keys
self.http = httplib2.Http() # For Basic Auth... self.opener = None
if self.username is not None and self.password is not None: if self.username is not None and self.password is not None:
if self.authtype == "OAuth": if self.authtype == "OAuth":
pass pass
elif self.authtype == "Basic": elif self.authtype == "Basic":
self.http.add_credentials(self.username, self.password) self.auth_manager = urllib2.HTTPPasswordMgrWithDefaultRealm()
self.authenticated = True self.auth_manager.add_password(None, "http://twitter.com/account/verify_credentials.json", self.username, self.password)
else: self.handler = urllib2.HTTPBasicAuthHandler(self.auth_manager)
pass self.opener = urllib2.build_opener(self.handler)
self.authenticated = True
else:
pass
def shortenURL(self, url_to_shorten): def shortenURL(self, url_to_shorten):
# Perhaps we should have fallbacks here in case the is.gd API limit gets hit? Maybe allow them to set the host? # Perhaps we should have fallbacks here in case the is.gd API limit gets hit? Maybe allow them to set the host?
shortURL = urllib2.urlopen("http://is.gd/api.php?" + urllib.urlencode({"longurl": url_to_shorten})).read() shortURL = urllib2.urlopen("http://is.gd/api.php?" + urllib.urlencode({"longurl": url_to_shorten})).read()
return shortURL return shortURL
def constructApiURL(self, base_url, params): def constructApiURL(self, base_url, params):
queryURL = base_url queryURL = base_url
questionMarkUsed = False questionMarkUsed = False
for param in params: for param in params:
if params[param] is not None: if params[param] is not None:
queryURL += (("&" if questionMarkUsed is True else "?") + param + "=" + params[param]) queryURL += (("&" if questionMarkUsed is True else "?") + param + "=" + params[param])
questionMarkUsed = True questionMarkUsed = True
return queryURL return queryURL
def getPublicTimeline(self): def getPublicTimeline(self):
publicTimeline = simplejson.load(urllib2.urlopen("http://twitter.com/statuses/public_timeline.json")) publicTimeline = simplejson.load(urllib2.urlopen("http://twitter.com/statuses/public_timeline.json"))
formattedTimeline = [] formattedTimeline = []
for tweet in publicTimeline: for tweet in publicTimeline:
formattedTimeline.append(tweet['text']) formattedTimeline.append(tweet['text'])
return formattedTimeline return formattedTimeline
def getUserTimeline(self, **kwargs): def getUserTimeline(self, **kwargs):
# 99% API compliant, I think - need to figure out Gzip compression and auto-getting based on authentication # 99% API compliant, I think - need to figure out Gzip compression and auto-getting based on authentication
# By doing this with kwargs and constructing a url outside, we can stay somewhat agnostic of API changes - it's all # By doing this with kwargs and constructing a url outside, we can stay somewhat agnostic of API changes - it's all
# based on what the user decides to pass. We just handle the heavy lifting! :D # based on what the user decides to pass. We just handle the heavy lifting! :D
userTimelineURL = self.constructApiURL("http://twitter.com/statuses/user_timeline/" + self.username + ".json", kwargs) userTimelineURL = self.constructApiURL("http://twitter.com/statuses/user_timeline/" + self.username + ".json", kwargs)
userTimeline = simplejson.load(urllib2.urlopen(userTimelineURL)) userTimeline = simplejson.load(urllib2.urlopen(userTimelineURL))
formattedTimeline = [] formattedTimeline = []
for tweet in userTimeline: for tweet in userTimeline:
formattedTimeline.append(tweet['text']) formattedTimeline.append(tweet['text'])
return formattedTimeline return formattedTimeline
def getUserMentions(self, **kwargs): def getUserMentions(self, **kwargs):
if self.authenticated is True: if self.authenticated is True:
if self.authtype == "Basic": if self.authtype == "Basic":
pass pass
else: else:
pass pass
else: else:
print "getUserMentions() requires you to be authenticated." print "getUserMentions() requires you to be authenticated."
pass pass
def updateStatus(self, status = None, in_reply_to_status_id = None): def updateStatus(self, status, in_reply_to_status_id = ""):
if self.authenticated is True: if self.authenticated is True:
if self.authtype == "Basic": if self.authtype == "Basic":
self.http.request("http://twitter.com/statuses/update.json", "POST", urllib.urlencode({"status": status}, {"in_reply_to_status_id": in_reply_to_status_id})) self.opener.open("http://twitter.com/statuses/update.json" + urllib.urlencode({"status": status}, {"in_reply_to_status_id": in_reply_to_status_id}))
else: print self.opener.open("http://twitter.com/statuses/update.json" + urllib.urlencode({"status": status}, {"in_reply_to_status_id": in_reply_to_status_id})).read()
print "Sorry, OAuth support is still forthcoming. Feel free to help out on this front!" else:
pass print "Sorry, OAuth support is still forthcoming. Feel free to help out on this front!"
else: pass
print "updateStatus() requires you to be authenticated." else:
pass print "updateStatus() requires you to be authenticated."
pass
def destroyStatus(self, id): def destroyStatus(self, id):
if self.authenticated is True: if self.authenticated is True:
self.http.request("http://twitter.com/status/destroy/" + id + ".json", "POST") self.http.request("http://twitter.com/status/destroy/" + id + ".json", "POST")
else: else:
print "destroyStatus() requires you to be authenticated." print "destroyStatus() requires you to be authenticated."
pass pass
def getSearchTimeline(self, search_query, optional_page): def getSearchTimeline(self, search_query, optional_page):
params = urllib.urlencode({'q': search_query, 'rpp': optional_page}) # Doesn't hurt to do pages this way. *shrug* params = urllib.urlencode({'q': search_query, 'rpp': optional_page}) # Doesn't hurt to do pages this way. *shrug*
searchTimeline = simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params)) searchTimeline = simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params))
formattedTimeline = [] formattedTimeline = []
for tweet in searchTimeline['results']: for tweet in searchTimeline['results']:
formattedTimeline.append(tweet['text']) formattedTimeline.append(tweet['text'])
return formattedTimeline return formattedTimeline
def getCurrentTrends(self): def getCurrentTrends(self):
# Returns an array of dictionary items containing the current trends # Returns an array of dictionary items containing the current trends
trendingTopicsURL = "http://search.twitter.com/trends.json" trendingTopicsURL = "http://search.twitter.com/trends.json"
trendingTopics = simplejson.load(urllib.urlopen(trendingTopicsURL)) trendingTopics = simplejson.load(urllib.urlopen(trendingTopicsURL))
trendingTopicsArray = [] trendingTopicsArray = []
for topic in trendingTopics['trends']: for topic in trendingTopics['trends']:
trendingTopicsArray.append({"name" : topic['name'], "url" : topic['url']}) trendingTopicsArray.append({"name" : topic['name'], "url" : topic['url']})
return trendingTopicsArray return trendingTopicsArray