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

All methods now return a Python/JSON value depicting the data Twitter sends back

Browse source
This commit is contained in:
Ryan McGrath 2009-06-08 01:26:32 -04:00
parent 5342fbccc6
commit 402a2dddc0
1 changed files with 44 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

113
tango.py
View file

@ -10,7 +10,7 @@
Questions, comments? ryan@venodesigns.net Questions, comments? ryan@venodesigns.net
""" """
import httplib, urllib, urllib2, mimetypes, mimetools import httplib, urllib, urllib2, mimetypes, mimetools, pprint
from urllib2 import HTTPError from urllib2 import HTTPError
@ -41,8 +41,6 @@ class setup:
self.handler = urllib2.HTTPBasicAuthHandler(self.auth_manager) self.handler = urllib2.HTTPBasicAuthHandler(self.auth_manager)
self.opener = urllib2.build_opener(self.handler) self.opener = urllib2.build_opener(self.handler)
self.authenticated = True self.authenticated = True
else:
pass
# OAuth functions; shortcuts for verifying the credentials. # OAuth functions; shortcuts for verifying the credentials.
def fetch_response_oauth(self, oauth_request): def fetch_response_oauth(self, oauth_request):
@ -59,8 +57,12 @@ class setup:
# URL Shortening function huzzah # URL Shortening function huzzah
def shortenURL(self, url_to_shorten, shortener = "http://is.gd/api.php", query = "longurl"): def shortenURL(self, url_to_shorten, shortener = "http://is.gd/api.php", query = "longurl"):
shortURL = urllib2.urlopen(shortener + "?" + urllib.urlencode({query: url_to_shorten})).read() try:
return shortURL return urllib2.urlopen(shortener + "?" + urllib.urlencode({query: url_to_shorten})).read()
except HTTPError, e:
if self.debug is True:
print e.headers
print "shortenURL() failed with a " + str(e.code) + " error code."
def constructApiURL(self, base_url, params): def constructApiURL(self, base_url, params):
queryURL = base_url queryURL = base_url
@ -71,38 +73,20 @@ class setup:
questionMarkUsed = True questionMarkUsed = True
return queryURL return queryURL
def createGenericTimeline(self, existingTimeline):
# Many of Twitters API functions return the same style of data. This function just wraps it if we need it.
genericTimeline = []
for tweet in existingTimeline:
genericTimeline.append({
"created_at": tweet["created_at"],
"in_reply_to_screen_name": tweet["in_reply_to_screen_name"],
"in_reply_to_status_id": tweet["in_reply_to_status_id"],
"in_reply_to_user_id": tweet["in_reply_to_user_id"],
"id": tweet["id"],
"text": tweet["text"]
})
return genericTimeline
def getRateLimitStatus(self, rate_for = "requestingIP"): def getRateLimitStatus(self, rate_for = "requestingIP"):
try: try:
if rate_for == "requestingIP": if rate_for == "requestingIP":
rate_limit = simplejson.load(urllib2.urlopen("http://twitter.com/account/rate_limit_status.json")) return simplejson.load(urllib2.urlopen("http://twitter.com/account/rate_limit_status.json"))
else: else:
rate_limit = simplejson.load(self.opener.open("http://twitter.com/account/rate_limit_status.json")) return simplejson.load(self.opener.open("http://twitter.com/account/rate_limit_status.json"))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
print "It seems that there's something wrong. Twitter gave you a " + str(e.code) + " error code; are you doing something you shouldn't be?" print "It seems that there's something wrong. Twitter gave you a " + str(e.code) + " error code; are you doing something you shouldn't be?"
return {"remaining-hits": rate_limit["remaining-hits"],
"hourly-limit": rate_limit["hourly-limit"],
"reset-time": rate_limit["reset-time"],
"reset-time-in-seconds": rate_limit["reset-time-in-seconds"]}
def getPublicTimeline(self): def getPublicTimeline(self):
try: try:
return self.createGenericTimeline(simplejson.load(urllib2.urlopen("http://twitter.com/statuses/public_timeline.json"))) return simplejson.load(urllib2.urlopen("http://twitter.com/statuses/public_timeline.json"))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -112,7 +96,7 @@ class setup:
if self.authenticated is True: if self.authenticated is True:
try: try:
friendsTimelineURL = self.constructApiURL("http://twitter.com/statuses/friends_timeline.json", kwargs) friendsTimelineURL = self.constructApiURL("http://twitter.com/statuses/friends_timeline.json", kwargs)
return self.createGenericTimeline(simplejson.load(self.opener.open(friendsTimelineURL))) return simplejson.load(self.opener.open(friendsTimelineURL))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -124,7 +108,7 @@ class setup:
def getUserTimeline(self, **kwargs): def getUserTimeline(self, **kwargs):
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)
try: try:
return self.createGenericTimeline(simplejson.load(urllib2.urlopen(userTimelineURL))) return simplejson.load(urllib2.urlopen(userTimelineURL))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -135,8 +119,7 @@ class setup:
if self.authenticated is True: if self.authenticated is True:
try: try:
mentionsFeedURL = self.constructApiURL("http://twitter.com/statuses/mentions.json", kwargs) mentionsFeedURL = self.constructApiURL("http://twitter.com/statuses/mentions.json", kwargs)
mentionsFeed = simplejson.load(self.opener.open(mentionsFeedURL)) return simplejson.load(self.opener.open(mentionsFeedURL))
return self.createGenericTimeline(mentionsFeed)
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -147,13 +130,7 @@ class setup:
def showStatus(self, id): def showStatus(self, id):
try: try:
tweet = simplejson.load(self.opener.open("http://twitter.com/statuses/show/" + id + ".json")) return simplejson.load(self.opener.open("http://twitter.com/statuses/show/" + id + ".json"))
return {"created_at": tweet["created_at"],
"id": tweet["id"],
"text": tweet["text"],
"in_reply_to_status_id": tweet["in_reply_to_status_id"],
"in_reply_to_user_id": tweet["in_reply_to_user_id"],
"in_reply_to_screen_name": tweet["in_reply_to_screen_name"]}
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -163,7 +140,7 @@ class setup:
def updateStatus(self, status, in_reply_to_status_id = None): def updateStatus(self, status, in_reply_to_status_id = None):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/statuses/update.json?", urllib.urlencode({"status": status}, {"in_reply_to_status_id": in_reply_to_status_id})) return simplejson.load(self.opener.open("http://twitter.com/statuses/update.json?", urllib.urlencode({"status": status}, {"in_reply_to_status_id": in_reply_to_status_id})))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -175,7 +152,7 @@ class setup:
def destroyStatus(self, id): def destroyStatus(self, id):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/status/destroy/" + id + ".json", "POST") return simplejson.load(self.opener.open("http://twitter.com/status/destroy/" + id + ".json", "POST"))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -312,7 +289,7 @@ class setup:
def updateDeliveryDevice(self, device_name = "none"): def updateDeliveryDevice(self, device_name = "none"):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/account/update_delivery_device.json?", urllib.urlencode({"device": device_name})) return self.opener.open("http://twitter.com/account/update_delivery_device.json?", urllib.urlencode({"device": device_name}))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -323,7 +300,7 @@ class setup:
def updateProfileColors(self, **kwargs): def updateProfileColors(self, **kwargs):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open(self.constructApiURL("http://twitter.com/account/update_profile_colors.json?", kwargs)) return self.opener.open(self.constructApiURL("http://twitter.com/account/update_profile_colors.json?", kwargs))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -379,7 +356,7 @@ class setup:
if updateProfileQueryString != "": if updateProfileQueryString != "":
try: try:
self.opener.open("http://twitter.com/account/update_profile.json?", updateProfileQueryString) return self.opener.open("http://twitter.com/account/update_profile.json?", updateProfileQueryString)
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -391,8 +368,7 @@ class setup:
def getFavorites(self, page = "1"): def getFavorites(self, page = "1"):
if self.authenticated is True: if self.authenticated is True:
try: try:
favoritesTimeline = simplejson.load(self.opener.open("http://twitter.com/favorites.json?page=" + page)) return simplejson.load(self.opener.open("http://twitter.com/favorites.json?page=" + page))
return self.createGenericTimeline(favoritesTimeline)
except HTTPError, e: except HTTPError, e:
if self.debug: if self.debug:
print e.headers print e.headers
@ -403,7 +379,7 @@ class setup:
def createFavorite(self, id): def createFavorite(self, id):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/favorites/create/" + id + ".json", "") return simplejson.load(self.opener.open("http://twitter.com/favorites/create/" + id + ".json", ""))
except HTTPError, e: except HTTPError, e:
if self.debug: if self.debug:
print e.headers print e.headers
@ -414,7 +390,7 @@ class setup:
def destroyFavorite(self, id): def destroyFavorite(self, id):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/favorites/destroy/" + id + ".json", "") return simplejson.load(self.opener.open("http://twitter.com/favorites/destroy/" + id + ".json", ""))
except HTTPError, e: except HTTPError, e:
if self.debug: if self.debug:
print e.headers print e.headers
@ -432,7 +408,7 @@ class setup:
if screen_name is not None: if screen_name is not None:
apiURL = "http://twitter.com/notifications/follow/follow.json?screen_name=" + screen_name apiURL = "http://twitter.com/notifications/follow/follow.json?screen_name=" + screen_name
try: try:
self.opener.open(apiURL, "") return simplejson.load(self.opener.open(apiURL, ""))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -450,7 +426,7 @@ class setup:
if screen_name is not None: if screen_name is not None:
apiURL = "http://twitter.com/notifications/leave/leave.json?screen_name=" + screen_name apiURL = "http://twitter.com/notifications/leave/leave.json?screen_name=" + screen_name
try: try:
self.opener.open(apiURL, "") return simplejson.load(self.opener.open(apiURL, ""))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -491,7 +467,7 @@ class setup:
def createBlock(self, id): def createBlock(self, id):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/blocks/create/" + id + ".json", "") return simplejson.load(self.opener.open("http://twitter.com/blocks/create/" + id + ".json", ""))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -502,7 +478,7 @@ class setup:
def destroyBlock(self, id): def destroyBlock(self, id):
if self.authenticated is True: if self.authenticated is True:
try: try:
self.opener.open("http://twitter.com/blocks/destroy/" + id + ".json", "") return simplejson.load(self.opener.open("http://twitter.com/blocks/destroy/" + id + ".json", ""))
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -550,20 +526,7 @@ class setup:
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*
try: try:
searchTimeline = simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params)) return simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params))
# This one is custom built because the search feed is a bit different than the other feeds.
genericTimeline = []
for tweet in searchTimeline["results"]:
genericTimeline.append({
"created_at": tweet["created_at"],
"from_user": tweet["from_user"],
"from_user_id": tweet["from_user_id"],
"profile_image_url": tweet["profile_image_url"],
"id": tweet["id"],
"text": tweet["text"],
"to_user_id": tweet["to_user_id"]
})
return genericTimeline
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -571,13 +534,7 @@ class setup:
def getCurrentTrends(self): def getCurrentTrends(self):
try: try:
# Returns an array of dictionary items containing the current trends return simplejson.load(urllib.urlopen("http://search.twitter.com/trends.json"))
trendingTopicsURL = "http://search.twitter.com/trends.json"
trendingTopics = simplejson.load(urllib.urlopen(trendingTopicsURL))
trendingTopicsArray = []
for topic in trendingTopics['trends']:
trendingTopicsArray.append({"name" : topic['name'], "url" : topic['url']})
return trendingTopicsArray
except HTTPError, e: except HTTPError, e:
if self.debug is True: if self.debug is True:
print e.headers print e.headers
@ -586,15 +543,17 @@ class setup:
# The following methods are apart from the other Account methods, because they rely on a whole multipart-data posting function set. # The following methods are apart from the other Account methods, because they rely on a whole multipart-data posting function set.
def updateProfileBackgroundImage(self, filename, tile="true"): def updateProfileBackgroundImage(self, filename, tile="true"):
if self.authenticated is True: if self.authenticated is True:
#try: try:
files = [("image", filename, open(filename).read())] files = [("image", filename, open(filename).read())]
fields = [] fields = []
content_type, body = self.encode_multipart_formdata(fields, files) content_type, body = self.encode_multipart_formdata(fields, files)
headers = {'Content-Type': content_type, 'Content-Length': str(len(body))} headers = {'Content-Type': content_type, 'Content-Length': str(len(body))}
r = urllib2.Request("http://twitter.com/account/update_profile_background_image.json?tile=" + tile, body, headers) r = urllib2.Request("http://twitter.com/account/update_profile_background_image.json?tile=" + tile, body, headers)
return self.opener.open(r).read() return self.opener.open(r).read()
#except: except HTTPError, e:
#print "Oh god, this failed so horribly." if self.debug is True:
print e.headers
print "updateProfileBackgroundImage() failed with a " + str(e.code) + " error code."
else: else:
print "You realize you need to be authenticated to change a background image, right?" print "You realize you need to be authenticated to change a background image, right?"
@ -607,8 +566,10 @@ class setup:
headers = {'Content-Type': content_type, 'Content-Length': str(len(body))} headers = {'Content-Type': content_type, 'Content-Length': str(len(body))}
r = urllib2.Request("http://twitter.com/account/update_profile_image.json", body, headers) r = urllib2.Request("http://twitter.com/account/update_profile_image.json", body, headers)
return self.opener.open(r).read() return self.opener.open(r).read()
except: except HTTPError, e:
print "Oh god, this failed so horribly." if self.debug is True:
print e.headers
print "updateProfileImage() failed with a " + str(e.code) + " error code."
else: else:
print "You realize you need to be authenticated to change a profile image, right?" print "You realize you need to be authenticated to change a profile image, right?"