Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def getUserTimeline(self, id = None, **kwargs):
if id is not None and kwargs.has_key("user_id") is False and kwargs.has_key("screen_name") is False:
userTimelineURL = self.constructApiURL("http://twitter.com/statuses/user_timeline/" + id + ".json", kwargs)
elif id is None and kwargs.has_key("user_id") is False and kwargs.has_key("screen_name") is False and self.authenticated is True:
userTimelineURL = self.constructApiURL("http://twitter.com/statuses/user_timeline/" + self.username + ".json", kwargs)
else:
userTimelineURL = self.constructApiURL("http://twitter.com/statuses/user_timeline.json", kwargs)
try:
# We do our custom opener if we're authenticated, as it helps avoid cases where it's a protected user
if self.authenticated is True:
return simplejson.load(self.opener.open(userTimelineURL))
else:
return simplejson.load(urllib2.urlopen(userTimelineURL))
except HTTPError, e:
raise TangoError("Failed with a %s error code. Does this user hide/protect their updates? If so, you'll need to authenticate and be their friend to get their timeline."
% `e.code`, e.code)
if description is not None:
if len(list(description)) < 160:
if useAmpersands is True:
updateProfileQueryString += "&" + urllib.urlencode({"description": description})
else:
updateProfileQueryString += urllib.urlencode({"description": description})
else:
raise TangoError("Twitter has a character limit of 160 for all descriptions. Try again.")
if updateProfileQueryString != "":
try:
return self.opener.open("http://twitter.com/account/update_profile.json?", updateProfileQueryString)
except HTTPError, e:
raise TangoError("updateProfile() failed with a %s error code." % `e.code`, e.code)
else:
raise TangoError("updateProfile() requires you to be authenticated.")
self.access_token = None
# Check and set up authentication
if self.username is not None and self.password is not None:
if self.authtype == "Basic":
# Basic authentication ritual
self.auth_manager = urllib2.HTTPPasswordMgrWithDefaultRealm()
self.auth_manager.add_password(None, "http://twitter.com", self.username, self.password)
self.handler = urllib2.HTTPBasicAuthHandler(self.auth_manager)
self.opener = urllib2.build_opener(self.handler)
if headers is not None:
self.opener.addheaders = [('User-agent', headers)]
try:
simplejson.load(self.opener.open("http://twitter.com/account/verify_credentials.json"))
self.authenticated = True
except HTTPError, e:
raise TangoError("Authentication failed with your provided credentials. Try again? (%s failure)" % `e.code`, e.code)
else:
self.signature_method = oauth.OAuthSignatureMethod_HMAC_SHA1()
# Awesome OAuth authentication ritual
if consumer_secret is not None and consumer_key is not None:
#req = oauth.OAuthRequest.from_consumer_and_token
#req.sign_request(self.signature_method, self.consumer_key, self.token)
#self.opener = urllib2.build_opener()
pass
else:
raise TwythonError("Woah there, buddy. We've defaulted to OAuth authentication, but you didn't provide API keys. Try again.")
def createFriendship(self, id = None, user_id = None, screen_name = None, follow = "false"):
if self.authenticated is True:
apiURL = ""
if id is not None:
apiURL = "http://twitter.com/friendships/create/" + id + ".json" + "?follow=" + follow
if user_id is not None:
apiURL = "http://twitter.com/friendships/create.json?user_id=" + user_id + "&follow=" + follow
if screen_name is not None:
apiURL = "http://twitter.com/friendships/create.json?screen_name=" + screen_name + "&follow=" + follow
try:
return simplejson.load(self.opener.open(apiURL))
except HTTPError, e:
# Rate limiting is done differently here for API reasons...
if e.code == 403:
raise TangoError("You've hit the update limit for this method. Try again in 24 hours.")
raise TangoError("createFriendship() failed with a %s error code." % `e.code`, e.code)
else:
raise TangoError("createFriendship() requires you to be authenticated.")
def getFriendsTimeline(self, **kwargs):
if self.authenticated is True:
try:
friendsTimelineURL = self.constructApiURL("http://twitter.com/statuses/friends_timeline.json", kwargs)
return simplejson.load(self.opener.open(friendsTimelineURL))
except HTTPError, e:
raise TangoError("getFriendsTimeline() failed with a %s error code." % `e.code`)
else:
raise TangoError("getFriendsTimeline() requires you to be authenticated.")
def updateDeliveryDevice(self, device_name = "none"):
if self.authenticated is True:
try:
return self.opener.open("http://twitter.com/account/update_delivery_device.json?", urllib.urlencode({"device": device_name}))
except HTTPError, e:
raise TangoError("updateDeliveryDevice() failed with a %s error code." % `e.code`, e.code)
else:
raise TangoError("updateDeliveryDevice() requires you to be authenticated.")
if useAmpersands is True:
updateProfileQueryString += "&" + urllib.urlencode({"url": url})
else:
updateProfileQueryString += urllib.urlencode({"url": url})
useAmpersands = True
else:
raise TangoError("Twitter has a character limit of 100 for all urls. Try again.")
if location is not None:
if len(list(location)) < 30:
if useAmpersands is True:
updateProfileQueryString += "&" + urllib.urlencode({"location": location})
else:
updateProfileQueryString += urllib.urlencode({"location": location})
useAmpersands = True
else:
raise TangoError("Twitter has a character limit of 30 for all locations. Try again.")
if description is not None:
if len(list(description)) < 160:
if useAmpersands is True:
updateProfileQueryString += "&" + urllib.urlencode({"description": description})
else:
updateProfileQueryString += urllib.urlencode({"description": description})
else:
raise TangoError("Twitter has a character limit of 160 for all descriptions. Try again.")
if updateProfileQueryString != "":
try:
return self.opener.open("http://twitter.com/account/update_profile.json?", updateProfileQueryString)
except HTTPError, e:
raise TangoError("updateProfile() failed with a %s error code." % `e.code`, e.code)
else:
raise TangoError("updateProfile() requires you to be authenticated.")
def searchTwitter(self, search_query, **kwargs):
searchURL = self.constructApiURL("http://search.twitter.com/search.json", kwargs) + "&" + urllib.urlencode({"q": search_query})
try:
return simplejson.load(urllib2.urlopen(searchURL))
except HTTPError, e:
raise TangoError("getSearchTimeline() failed with a %s error code." % `e.code`, e.code)
def getSentMessages(self, since_id = None, max_id = None, count = None, page = "1"):
if self.authenticated is True:
apiURL = "http://twitter.com/direct_messages/sent.json?page=" + page
if since_id is not None:
apiURL += "&since_id=" + since_id
if max_id is not None:
apiURL += "&max_id=" + max_id
if count is not None:
apiURL += "&count=" + count
try:
return simplejson.load(self.opener.open(apiURL))
except HTTPError, e:
raise TangoError("getSentMessages() failed with a %s error code." % `e.code`, e.code)
else:
raise TangoError("getSentMessages() requires you to be authenticated.")
def showStatus(self, id):
try:
if self.authenticated is True:
return simplejson.load(self.opener.open("http://twitter.com/statuses/show/%s.json" % id))
else:
return simplejson.load(urllib2.urlopen("http://twitter.com/statuses/show/%s.json" % id))
except HTTPError, e:
raise TangoError("Failed with a %s error code. Does this user hide/protect their updates? You'll need to authenticate and be friends to get their timeline."
% `e.code`, e.code)