Python v3

twitter/_file_cache.py

@@ -19,7 +19,8 @@ def __init__(self, root_directory=None):
     def Get(self, key):
         path = self._GetPath(key)
         if os.path.exists(path):
-            return open(path).read()
+            with open(path) as f:
+                return f.read()
         else:
             return None
 
@@ -85,7 +86,7 @@ def _InitializeRootDirectory(self, root_directory):
 
     def _GetPath(self, key):
         try:
-            hashed_key = md5(key).hexdigest()
+            hashed_key = md5(key.encode('utf-8')).hexdigest()
         except TypeError:
             hashed_key = md5.new(key).hexdigest()
 

twitter/api.py

@@ -46,6 +46,13 @@
 from twitter import (__version__, _FileCache, simplejson, DirectMessage, List,
                      Status, Trend, TwitterError, User, UserStatus)
 
+try:
+  # python 3
+  urllib_version = urllib.request.__version__
+except AttributeError:
+  # python 2
+  urllib_version = urllib.__version__
+
 CHARACTER_LIMIT = 140
 
 # A singleton representing a lazily instantiated FileCache.
@@ -3468,7 +3475,7 @@ def _InitializeRequestHeaders(self, request_headers):
 
     def _InitializeUserAgent(self):
         user_agent = 'Python-urllib/%s (python-twitter/%s)' % \
-                     (urllib.__version__, __version__)
+                     (urllib_version, __version__)
         self.SetUserAgent(user_agent)
 
     def _InitializeDefaultParameters(self):

twitter/direct_message.py

@@ -1,8 +1,13 @@
-from builtins import object
 #!/usr/bin/env python
 
+from builtins import object
+
 from calendar import timegm
-import rfc822
+
+try:
+    from rfc822 import parsedate
+except ImportError:
+    from email.utils import parsedate
 
 from twitter import simplejson, TwitterError
 
@@ -107,7 +112,7 @@ def GetCreatedAtInSeconds(self):
         Returns:
           The time this direct message was posted, in seconds since the epoch.
         """
-        return timegm(rfc822.parsedate(self.created_at))
+        return timegm(parsedate(self.created_at))
 
     created_at_in_seconds = property(GetCreatedAtInSeconds,
                                      doc="The time this direct message was "

twitter/status.py

@@ -4,7 +4,12 @@
 from builtins import object
 from past.utils import old_div
 from calendar import timegm
-import rfc822
+
+try:
+    from rfc822 import parsedate
+except ImportError:
+    from email.utils import parsedate
+
 import time
 
 from twitter import simplejson, Hashtag, TwitterError, Url
@@ -152,7 +157,7 @@ def GetCreatedAtInSeconds(self):
         Returns:
           The time this status message was posted, in seconds since the epoch.
         """
-        return timegm(rfc822.parsedate(self.created_at))
+        return timegm(parsedate(self.created_at))
 
     created_at_in_seconds = property(GetCreatedAtInSeconds,
                                      doc="The time this status message was "

twitter_test.py

@@ -20,18 +20,22 @@
 
 __author__ = '[email protected]'
 
+from future import standard_library
+standard_library.install_aliases()
+
 import os
 import json as simplejson
 import time
 import calendar
 import unittest
-import urllib
+import urllib.request, urllib.parse, urllib.error
 
 import twitter
 
 class StatusTest(unittest.TestCase):
 
-  SAMPLE_JSON = '''{"created_at": "Fri Jan 26 23:17:14 +0000 2007", "id": 4391023, "text": "A l\u00e9gp\u00e1rn\u00e1s haj\u00f3m tele van angoln\u00e1kkal.", "user": {"description": "Canvas. JC Penny. Three ninety-eight.", "id": 718443, "location": "Okinawa, Japan", "name": "Kesuke Miyagi", "profile_image_url": "https://twitter.com/system/user/profile_image/718443/normal/kesuke.png", "screen_name": "kesuke", "url": "https://twitter.com/kesuke"}}'''
+  # this is aiming to test escaped text passes through okay - the python interpreter will ignore the \u because this is a raw literal
+  SAMPLE_JSON = r'''{"created_at": "Fri Jan 26 23:17:14 +0000 2007", "id": 4391023, "text": "A l\u00e9gp\u00e1rn\u00e1s haj\u00f3m tele van angoln\u00e1kkal.", "user": {"description": "Canvas. JC Penny. Three ninety-eight.", "id": 718443, "location": "Okinawa, Japan", "name": "Kesuke Miyagi", "profile_image_url": "https://twitter.com/system/user/profile_image/718443/normal/kesuke.png", "screen_name": "kesuke", "url": "https://twitter.com/kesuke"}}'''
 
   def _GetSampleUser(self):
     return twitter.User(id=718443,
@@ -315,7 +319,7 @@ class FileCacheTest(unittest.TestCase):
   def testInit(self):
     """Test the twitter._FileCache constructor"""
     cache = twitter._FileCache()
-    self.assert_(cache is not None, 'cache is None')
+    self.assertTrue(cache is not None, 'cache is None')
 
   def testSet(self):
     """Test the twitter._FileCache.Set method"""
@@ -346,7 +350,7 @@ def testGetCachedTime(self):
     cache.Set("foo",'Hello World!')
     cached_time = cache.GetCachedTime("foo")
     delta = cached_time - now
-    self.assert_(delta <= 1,
+    self.assertTrue(delta <= 1,
                  'Cached time differs from clock time by more than 1 second.')
     cache.Remove("foo")
 
@@ -362,7 +366,7 @@ def setUp(self):
                       cache=None)
     api.SetUrllib(self._urllib)
     self._api = api
-    print "Testing the API class. This test is time controlled"
+    print("Testing the API class. This test is time controlled")
 
   def testTwitterError(self):
     '''Test that twitter responses containing an error message are wrapped.'''
@@ -371,7 +375,7 @@ def testTwitterError(self):
     # Manually try/catch so we can check the exception's value
     try:
       statuses = self._api.GetUserTimeline()
-    except twitter.TwitterError, error:
+    except twitter.TwitterError as error:
       # If the error message matches, the test passes
       self.assertEqual('test error', error.message)
     else:
@@ -380,7 +384,7 @@ def testTwitterError(self):
   def testGetUserTimeline(self):
     '''Test the twitter.Api GetUserTimeline method'''
     time.sleep(8)
-    print 'Testing GetUserTimeline'
+    print('Testing GetUserTimeline')
     self._AddHandler('https://api.twitter.com/1.1/statuses/user_timeline.json?count=1&screen_name=kesuke',
                      curry(self._OpenTestData, 'user_timeline-kesuke.json'))
     statuses = self._api.GetUserTimeline(screen_name='kesuke', count=1)
@@ -400,7 +404,7 @@ def testGetUserTimeline(self):
   def testGetStatus(self):
     '''Test the twitter.Api GetStatus method'''
     time.sleep(8)
-    print 'Testing GetStatus'
+    print('Testing GetStatus')
     self._AddHandler('https://api.twitter.com/1.1/statuses/show.json?include_my_retweet=1&id=89512102',
                      curry(self._OpenTestData, 'show-89512102.json'))
     status = self._api.GetStatus(89512102)
@@ -410,7 +414,7 @@ def testGetStatus(self):
   def testDestroyStatus(self):
     '''Test the twitter.Api DestroyStatus method'''
     time.sleep(8)
-    print 'Testing DestroyStatus'
+    print('Testing DestroyStatus')
     self._AddHandler('https://api.twitter.com/1.1/statuses/destroy/103208352.json',
                      curry(self._OpenTestData, 'status-destroy.json'))
     status = self._api.DestroyStatus(103208352)
@@ -419,7 +423,7 @@ def testDestroyStatus(self):
   def testPostUpdate(self):
     '''Test the twitter.Api PostUpdate method'''
     time.sleep(8)
-    print 'Testing PostUpdate'
+    print('Testing PostUpdate')
     self._AddHandler('https://api.twitter.com/1.1/statuses/update.json',
                      curry(self._OpenTestData, 'update.json'))
     status = self._api.PostUpdate(u'Моё судно на воздушной подушке полно угрей'.encode('utf8'))
@@ -429,7 +433,7 @@ def testPostUpdate(self):
   def testPostRetweet(self):
     '''Test the twitter.Api PostRetweet method'''
     time.sleep(8)
-    print 'Testing PostRetweet'
+    print('Testing PostRetweet')
     self._AddHandler('https://api.twitter.com/1.1/statuses/retweet/89512102.json',
                      curry(self._OpenTestData, 'retweet.json'))
     status = self._api.PostRetweet(89512102)
@@ -438,20 +442,20 @@ def testPostRetweet(self):
   def testPostUpdateLatLon(self):
     '''Test the twitter.Api PostUpdate method, when used in conjunction with latitude and longitude'''
     time.sleep(8)
-    print 'Testing PostUpdateLatLon'
+    print('Testing PostUpdateLatLon')
     self._AddHandler('https://api.twitter.com/1.1/statuses/update.json',
                      curry(self._OpenTestData, 'update_latlong.json'))
     #test another update with geo parameters, again test somewhat arbitrary
     status = self._api.PostUpdate(u'Моё судно на воздушной подушке полно угрей'.encode('utf8'), latitude=54.2, longitude=-2)
     self.assertEqual(u'Моё судно на воздушной подушке полно угрей', status.text)
-    self.assertEqual(u'Point',status.GetGeo()['type'])
+    self.assertEqual('Point',status.GetGeo()['type'])
     self.assertEqual(26.2,status.GetGeo()['coordinates'][0])
     self.assertEqual(127.5,status.GetGeo()['coordinates'][1])
 
   def testGetReplies(self):
     '''Test the twitter.Api GetReplies method'''
     time.sleep(8)
-    print 'Testing GetReplies'
+    print('Testing GetReplies')
     self._AddHandler('https://api.twitter.com/1.1/statuses/user_timeline.json',
                      curry(self._OpenTestData, 'replies.json'))
     statuses = self._api.GetReplies()
@@ -460,7 +464,7 @@ def testGetReplies(self):
   def testGetRetweetsOfMe(self):
     '''Test the twitter.API GetRetweetsOfMe method'''
     time.sleep(8)
-    print 'Testing GetRetweetsOfMe'
+    print('Testing GetRetweetsOfMe')
     self._AddHandler('https://api.twitter.com/1.1/statuses/retweets_of_me.json',
         curry(self._OpenTestData, 'retweets_of_me.json'))
     retweets = self._api.GetRetweetsOfMe()
@@ -469,7 +473,7 @@ def testGetRetweetsOfMe(self):
   def testGetFriends(self):
     '''Test the twitter.Api GetFriends method'''
     time.sleep(8)
-    print 'Testing GetFriends'
+    print('Testing GetFriends')
     self._AddHandler('https://api.twitter.com/1.1/friends/list.json?cursor=123',
                      curry(self._OpenTestData, 'friends.json'))
     users = self._api.GetFriends(cursor=123)
@@ -479,7 +483,7 @@ def testGetFriends(self):
   def testGetFollowers(self):
     '''Test the twitter.Api GetFollowers method'''
     time.sleep(8)
-    print 'Testing GetFollowers'
+    print('Testing GetFollowers')
     self._AddHandler('https://api.twitter.com/1.1/followers/list.json?cursor=-1',
                      curry(self._OpenTestData, 'followers.json'))
     users = self._api.GetFollowers()
@@ -499,7 +503,7 @@ def testGetFollowers(self):
   def testGetDirectMessages(self):
     '''Test the twitter.Api GetDirectMessages method'''
     time.sleep(8)
-    print 'Testing GetDirectMessages'
+    print('Testing GetDirectMessages')
     self._AddHandler('https://api.twitter.com/1.1/direct_messages.json',
                      curry(self._OpenTestData, 'direct_messages.json'))
     statuses = self._api.GetDirectMessages()
@@ -508,7 +512,7 @@ def testGetDirectMessages(self):
   def testPostDirectMessage(self):
     '''Test the twitter.Api PostDirectMessage method'''
     time.sleep(8)
-    print 'Testing PostDirectMessage'
+    print('Testing PostDirectMessage')
     self._AddHandler('https://api.twitter.com/1.1/direct_messages/new.json',
                      curry(self._OpenTestData, 'direct_messages-new.json'))
     status = self._api.PostDirectMessage('test', u'Моё судно на воздушной подушке полно угрей'.encode('utf8'))
@@ -518,7 +522,7 @@ def testPostDirectMessage(self):
   def testDestroyDirectMessage(self):
     '''Test the twitter.Api DestroyDirectMessage method'''
     time.sleep(8)
-    print 'Testing DestroyDirectMessage'
+    print('Testing DestroyDirectMessage')
     self._AddHandler('https://api.twitter.com/1.1/direct_messages/destroy.json',
                      curry(self._OpenTestData, 'direct_message-destroy.json'))
     status = self._api.DestroyDirectMessage(3496342)
@@ -528,7 +532,7 @@ def testDestroyDirectMessage(self):
   def testCreateFriendship(self):
     '''Test the twitter.Api CreateFriendship method'''
     time.sleep(8)
-    print 'Testing CreateFriendship'
+    print('Testing CreateFriendship')
     self._AddHandler('https://api.twitter.com/1.1/friendships/create.json',
                      curry(self._OpenTestData, 'friendship-create.json'))
     user = self._api.CreateFriendship('dewitt')
@@ -538,7 +542,7 @@ def testCreateFriendship(self):
   def testDestroyFriendship(self):
     '''Test the twitter.Api DestroyFriendship method'''
     time.sleep(8)
-    print 'Testing Destroy Friendship'
+    print('Testing Destroy Friendship')
     self._AddHandler('https://api.twitter.com/1.1/friendships/destroy.json',
                      curry(self._OpenTestData, 'friendship-destroy.json'))
     user = self._api.DestroyFriendship('dewitt')
@@ -548,7 +552,7 @@ def testDestroyFriendship(self):
   def testGetUser(self):
     '''Test the twitter.Api GetUser method'''
     time.sleep(8)
-    print 'Testing GetUser'
+    print('Testing GetUser')
     self._AddHandler('https://api.twitter.com/1.1/users/show.json?user_id=dewitt',
                      curry(self._OpenTestData, 'show-dewitt.json'))
     user = self._api.GetUser('dewitt')
@@ -622,8 +626,8 @@ def open(self, url, data=None):
       self._opened = True
       return self._handlers[url]()
     else:
-      print url
-      print self._handlers
+      print(url)
+      print(self._handlers)
 
       raise Exception('Unexpected URL %s (Checked: %s)' % (url, self._handlers))