Skip to content

GitLab

Commit a00f378d authored by Josh Roesslein's avatar Josh Roesslein
Browse files

2 space -> 4 space indents

parent 4b126837
Hide whitespace changes
Inline Side-by-side
streamwatcher.py
View file @ a00f378d
......@@ -5,17 +5,18 @@ from getpass import getpass
import tweepy
class StreamWatcherListener(tweepy.StreamListener):
def on_status(self, status):
print status.text
def on_status(self, status):
print status.text
def on_error(self, status_code):
print 'An error has occured! Status code = %s' % status_code
return True # keep stream alive
def on_error(self, status_code):
print 'An error has occured! Status code = %s' % status_code
return True # keep stream alive
def on_timeout(self):
print 'Snoozing Zzzzzz'
def on_timeout(self):
print 'Snoozing Zzzzzz'
# Prompt for login credentials and setup stream object
username = raw_input('Twitter username: ')
......@@ -24,38 +25,37 @@ stream = tweepy.Stream(username, password, StreamWatcherListener())
# Prompt for mode of streaming and connect
while True:
mode = raw_input('Mode? [sample/filter] ')
if mode == 'sample':
stream.sample()
break
elif mode == 'filter':
follow_list = raw_input('Users to follow (comma separated): ').strip()
track_list = raw_input('Keywords to track (comma seperated): ').strip()
if follow_list:
follow_list = [u for u in follow_list.split(',')]
else:
follow_list = None
if track_list:
track_list = [k for k in track_list.split(',')]
mode = raw_input('Mode? [sample/filter] ')
if mode == 'sample':
stream.sample()
break
elif mode == 'filter':
follow_list = raw_input('Users to follow (comma separated): ').strip()
track_list = raw_input('Keywords to track (comma seperated): ').strip()
if follow_list:
follow_list = [u for u in follow_list.split(',')]
else:
follow_list = None
if track_list:
track_list = [k for k in track_list.split(',')]
else:
track_list = None
stream.filter(follow_list, track_list)
break
else:
track_list = None
stream.filter(follow_list, track_list)
break
else:
print 'Invalid choice! Try again.'
print 'Invalid choice! Try again.'
# Run in a loop until termination
while True:
try:
if stream.running is False:
print 'Stream stopped!'
break
time.sleep(1)
except KeyboardInterrupt:
break
try:
if stream.running is False:
print 'Stream stopped!'
break
time.sleep(1)
except KeyboardInterrupt:
break
# Shutdown connection
stream.disconnect()
print 'Bye!'
tests.py
View file @ a00f378d
......@@ -5,177 +5,180 @@
import unittest
import random
from time import sleep
import os
from tweepy import *
"""Configurations"""
# Must supply twitter account credentials for tests
username = ''
password = ''
username = 'tweebly'
password = 'josh1987twitter'
"""Unit tests"""
# API tests
class TweepyAPITests(unittest.TestCase):
def setUp(self):
self.api = API(BasicAuthHandler(username, password))
def testpublictimeline(self):
self.assertEqual(len(self.api.public_timeline()), 20)
def testfriendstimeline(self):
self.assert_(len(self.api.friends_timeline()) > 0)
def testusertimeline(self):
s = self.api.user_timeline(screen_name='twitter')
self.assert_(len(s) > 0)
self.assertEqual(s[0].author.screen_name, 'twitter')
def testmentions(self):
s = self.api.mentions()
self.assert_(len(s) > 0)
self.assert_(s[0].text.find(username) >= 0)
def testgetstatus(self):
s = self.api.get_status(id=123)
self.assertEqual(s.author.id, 17)
def testupdateanddestroystatus(self):
# test update
text = 'testing %i' % random.randint(0,1000)
update = self.api.update_status(status=text)
self.assertEqual(update.text, text)
# test destroy
deleted = self.api.destroy_status(id=update.id)
self.assertEqual(deleted.id, update.id)
def testgetuser(self):
u = self.api.get_user(screen_name='twitter')
self.assertEqual(u.screen_name, 'twitter')
def testme(self):
me = self.api.me()
self.assertEqual(me.screen_name, username)
def testfriends(self):
friends = self.api.friends()
self.assert_(len(friends) > 0)
def testfollowers(self):
followers = self.api.followers()
self.assert_(len(followers) > 0)
def testdirectmessages(self):
dms = self.api.direct_messages()
self.assert_(len(dms) > 0)
def testsendanddestroydirectmessage(self):
# send
sent_dm = self.api.send_direct_message(username, 'test message')
self.assertEqual(sent_dm.text, 'test message')
self.assertEqual(sent_dm.sender.screen_name, username)
self.assertEqual(sent_dm.recipient.screen_name, username)
# destroy
destroyed_dm = self.api.destroy_direct_message(sent_dm.id)
self.assertEqual(destroyed_dm.text, sent_dm.text)
self.assertEqual(destroyed_dm.id, sent_dm.id)
self.assertEqual(destroyed_dm.sender.screen_name, username)
self.assertEqual(destroyed_dm.recipient.screen_name, username)
def testcreatefriendship(self):
friend = self.api.create_friendship('twitter')
self.assertEqual(friend.screen_name, 'twitter')
self.assertTrue(self.api.exists_friendship(username, 'twitter'))
def testdestroyfriendship(self):
enemy = self.api.destroy_friendship('twitter')
self.assertEqual(enemy.screen_name, 'twitter')
self.assertFalse(self.api.exists_friendship(username, 'twitter'))
def testshowfriendship(self):
source, target = self.api.show_friendship(target_screen_name='twtiter')
self.assert_(isinstance(source, Friendship))
self.assert_(isinstance(target, Friendship))
# Authentication tests
def setUp(self):
self.api = API(BasicAuthHandler(username, password))
def testpublictimeline(self):
self.assertEqual(len(self.api.public_timeline()), 20)
def testfriendstimeline(self):
self.assert_(len(self.api.friends_timeline()) > 0)
def testusertimeline(self):
s = self.api.user_timeline(screen_name='twitter')
self.assert_(len(s) > 0)
self.assertEqual(s[0].author.screen_name, 'twitter')
def testmentions(self):
s = self.api.mentions()
self.assert_(len(s) > 0)
self.assert_(s[0].text.find(username) >= 0)
def testgetstatus(self):
s = self.api.get_status(id=123)
self.assertEqual(s.author.id, 17)
def testupdateanddestroystatus(self):
# test update
text = 'testing %i' % random.randint(0, 1000)
update = self.api.update_status(status=text)
self.assertEqual(update.text, text)
# test destroy
deleted = self.api.destroy_status(id=update.id)
self.assertEqual(deleted.id, update.id)
def testgetuser(self):
u = self.api.get_user(screen_name='twitter')
self.assertEqual(u.screen_name, 'twitter')
def testme(self):
me = self.api.me()
self.assertEqual(me.screen_name, username)
def testfriends(self):
friends = self.api.friends()
self.assert_(len(friends) > 0)
def testfollowers(self):
followers = self.api.followers()
self.assert_(len(followers) > 0)
def testdirectmessages(self):
dms = self.api.direct_messages()
self.assert_(len(dms) > 0)
def testsendanddestroydirectmessage(self):
# send
sent_dm = self.api.send_direct_message(username, 'test message')
self.assertEqual(sent_dm.text, 'test message')
self.assertEqual(sent_dm.sender.screen_name, username)
self.assertEqual(sent_dm.recipient.screen_name, username)
# destroy
destroyed_dm = self.api.destroy_direct_message(sent_dm.id)
self.assertEqual(destroyed_dm.text, sent_dm.text)
self.assertEqual(destroyed_dm.id, sent_dm.id)
self.assertEqual(destroyed_dm.sender.screen_name, username)
self.assertEqual(destroyed_dm.recipient.screen_name, username)
def testcreatefriendship(self):
friend = self.api.create_friendship('twitter')
self.assertEqual(friend.screen_name, 'twitter')
self.assertTrue(self.api.exists_friendship(username, 'twitter'))
def testdestroyfriendship(self):
enemy = self.api.destroy_friendship('twitter')
self.assertEqual(enemy.screen_name, 'twitter')
self.assertFalse(self.api.exists_friendship(username, 'twitter'))
def testshowfriendship(self):
source, target = self.api.show_friendship(target_screen_name='twtiter')
self.assert_(isinstance(source, Friendship))
self.assert_(isinstance(target, Friendship))
class TweepyAuthTests(unittest.TestCase):
consumer_key = 'ZbzSsdQj7t68VYlqIFvdcA'
consumer_secret = '4yDWgrBiRs2WIx3bfvF9UWCRmtQ2YKpKJKBahtZcU'
consumer_key = 'ZbzSsdQj7t68VYlqIFvdcA'
consumer_secret = '4yDWgrBiRs2WIx3bfvF9UWCRmtQ2YKpKJKBahtZcU'
def testoauth(self):
auth = OAuthHandler(self.consumer_key, self.consumer_secret)
def testoauth(self):
auth = OAuthHandler(self.consumer_key, self.consumer_secret)
# test getting access token
auth_url = auth.get_authorization_url()
self.assert_(auth_url.startswith('http://twitter.com/oauth/authorize?'))
print 'Please authorize: ' + auth_url
verifier = raw_input('PIN: ').strip()
self.assert_(len(verifier) > 0)
access_token = auth.get_access_token(verifier)
self.assert_(access_token is not None)
# test getting access token
auth_url = auth.get_authorization_url()
self.assert_(auth_url.startswith('http://twitter.com/oauth/authorize?'))
print 'Please authorize: ' + auth_url
verifier = raw_input('PIN: ').strip()
self.assert_(len(verifier) > 0)
access_token = auth.get_access_token(verifier)
self.assert_(access_token is not None)
# build api object test using oauth
api = API(auth)
api.update_status('test %i' % random.randint(0,1000))
# build api object test using oauth
api = API(auth)
api.update_status('test %i' % random.randint(0, 1000))
def testbasicauth(self):
auth = BasicAuthHandler(username, password)
def testbasicauth(self):
auth = BasicAuthHandler(username, password)
# test accessing twitter API
api = API(auth)
api.update_status('test %i' % random.randint(1,1000))
# test accessing twitter API
api = API(auth)
api.update_status('test %i' % random.randint(1, 1000))
# Cache tests
class TweepyCacheTests(unittest.TestCase):
timeout = 2.0
memcache_servers = ['127.0.0.1:11211'] # must be running for test to pass
def _run_tests(self, do_cleanup=True):
# test store and get
self.cache.store('testkey', 'testvalue')
self.assertEqual(self.cache.get('testkey'), 'testvalue', 'Stored value does not match retrieved value')
# test timeout
sleep(self.timeout)
self.assertEqual(self.cache.get('testkey'), None, 'Cache entry should have expired')
# test cleanup
if do_cleanup:
self.cache.store('testkey', 'testvalue')
sleep(self.timeout)
self.cache.cleanup()
self.assertEqual(self.cache.count(), 0, 'Cache cleanup failed')
# test count
for i in range(0,20):
self.cache.store('testkey%i' % i, 'testvalue')
self.assertEqual(self.cache.count(), 20, 'Count is wrong')
# test flush
self.cache.flush()
self.assertEqual(self.cache.count(), 0, 'Cache failed to flush')
def testmemorycache(self):
self.cache = MemoryCache(timeout=self.timeout)
self._run_tests()
def testfilecache(self):
os.mkdir('cache_test_dir')
self.cache = FileCache('cache_test_dir', self.timeout)
self._run_tests()
self.cache.flush()
os.rmdir('cache_test_dir')
def testmemcache(self):
self.cache = MemCache(self.memcache_servers, self.timeout)
self._run_tests(do_cleanup=False)
timeout = 2.0
memcache_servers = ['127.0.0.1:11211'] # must be running for test to pass
def _run_tests(self, do_cleanup=True):
# test store and get
self.cache.store('testkey', 'testvalue')
self.assertEqual(self.cache.get('testkey'), 'testvalue',
'Stored value does not match retrieved value')
# test timeout
sleep(self.timeout)
self.assertEqual(self.cache.get('testkey'), None,
'Cache entry should have expired')
# test cleanup
if do_cleanup:
self.cache.store('testkey', 'testvalue')
sleep(self.timeout)
self.cache.cleanup()
self.assertEqual(self.cache.count(), 0, 'Cache cleanup failed')
# test count
for i in range(0, 20):
self.cache.store('testkey%i' % i, 'testvalue')
self.assertEqual(self.cache.count(), 20, 'Count is wrong')
# test flush
self.cache.flush()
self.assertEqual(self.cache.count(), 0, 'Cache failed to flush')
def testmemorycache(self):
self.cache = MemoryCache(timeout=self.timeout)
self._run_tests()
def testfilecache(self):
os.mkdir('cache_test_dir')
self.cache = FileCache('cache_test_dir', self.timeout)
self._run_tests()
self.cache.flush()
os.rmdir('cache_test_dir')
def testmemcache(self):
self.cache = MemCache(self.memcache_servers, self.timeout)
self._run_tests(do_cleanup=False)
if __name__ == '__main__':
unittest.main()
unittest.main()
tutorial/t2.py
View file @ a00f378d
......@@ -21,8 +21,8 @@ Let's query the public timeline and print it to the console...
public_timeline = no_auth_api.public_timeline()
print 'Public timeline...'
for status in public_timeline:
print status.text
print 'from: %s' % status.author.screen_name
print status.text
print 'from: %s' % status.author.screen_name
"""
Tweepy provides a non-authenticated instance of the API for you already
......@@ -50,8 +50,8 @@ and print it to the console...
friends_timeline = auth_api.friends_timeline()
print 'Friends timeline...'
for status in friends_timeline:
print status.text
print 'from: %s' % status.author.screen_name
print status.text
print 'from: %s' % status.author.screen_name
""" The End
......
tutorial/t3.py
View file @ a00f378d
......@@ -22,9 +22,9 @@ First let's create our own implementation of Status.
"""
class MyStatus(tweepy.Status):
def length(self):
"""Return length of status text"""
return len(self.text)
def length(self):
"""Return length of status text"""
return len(self.text)
"""
We must now register our implementation of Status with tweepy.
......@@ -54,10 +54,10 @@ to make sure data is present which your application depends on.
Here's a demo...
"""
try:
u = tweepy.api.get_user('twitter')
u = tweepy.api.get_user('twitter')
except TweepError, e:
# will be raised if user is invalid OR request failed
print 'Failed to get user: %s' % e
# will be raised if user is invalid OR request failed
print 'Failed to get user: %s' % e
"""
To disable auto validation...
......@@ -74,7 +74,7 @@ friends by using the User model friends() shortcut...
u = tweepy.api.get_user('twitter')
friends = u.friends()
for friend in friends:
print friend.screen_name
print friend.screen_name
"""
To learn about all shortcuts check out the reference documentation.
......
tutorial/t4.py
View file @ a00f378d
......@@ -14,9 +14,9 @@ When ever something goes wrong this exception will be raised.
Here is an example:
"""
try:
tweepy.api.update_status('this will fail since we are not authenticated!')
tweepy.api.update_status('this will fail since we are not authenticated!')
except tweepy.TweepError, e:
print 'Failed to update! %s' % e
print 'Failed to update! %s' % e
"""
TweepError's can be casted to string format which will
......
tweepy/__init__.py
View file @ a00f378d
......@@ -16,3 +16,4 @@ from . streaming import Stream, StreamListener
# Global, unauthenticated instance of API
api = API()
tweepy/api.py
View file @ a00f378d
......@@ -10,455 +10,460 @@ from . error import TweepError
from . auth import BasicAuthHandler, OAuthHandler
from tweepy.parsers import *
"""Twitter API"""
class API(object):
"""Twitter API"""
def __init__(self, auth_handler=None, host='twitter.com', cache=None,
secure=False, api_root='', validate=True):
# you may access these freely
self.auth_handler = auth_handler
self.host = host
self.api_root = api_root
self.cache = cache
self.secure = secure
self.validate = validate
# not a good idea to touch these
self._username = None
@staticmethod
def new(auth='basic', *args, **kargs):
if auth == 'basic':
return API(BasicAuthHandler(*args, **kargs))
elif auth == 'oauth':
return API(OAuthHandler(*args, **kargs))
else:
raise TweepError('Invalid auth type')
"""Get public timeline"""
public_timeline = bind_api(
path = '/statuses/public_timeline.json',
parser = parse_statuses,
allowed_param = []
)
"""Get friends timeline"""
friends_timeline = bind_api(
path = '/statuses/friends_timeline.json',
parser = parse_statuses,
allowed_param = ['since_id', 'max_id', 'count', 'page'],
require_auth = True
)
"""Get user timeline"""
user_timeline = bind_api(
path = '/statuses/user_timeline.json',
parser = parse_statuses,
allowed_param = ['id', 'user_id', 'screen_name', 'since_id',
'max_id', 'count', 'page']
)
"""Get mentions"""
mentions = bind_api(
path = '/statuses/mentions.json',
parser = parse_statuses,
allowed_param = ['since_id', 'max_id', 'count', 'page'],
require_auth = True
)
"""Show status"""
get_status = bind_api(
path = '/statuses/show.json',
parser = parse_status,
allowed_param = ['id']
)
"""Update status"""
update_status = bind_api(
path = '/statuses/update.json',
method = 'POST',
parser = parse_status,
allowed_param = ['status', 'in_reply_to_status_id'],
require_auth = True
)
"""Destroy status"""
destroy_status = bind_api(
path = '/statuses/destroy.json',
method = 'DELETE',
parser = parse_status,
allowed_param = ['id'],
require_auth = True
)
"""Show user"""
get_user = bind_api(
path = '/users/show.json',
parser = parse_user,
allowed_param = ['id', 'user_id', 'screen_name']
)
"""Get authenticated user"""
def me(self):
# if username not fetched, go get it...
if self._username is None:
if self.auth_handler is None:
raise TweepError('Authentication required')
try:
user = bind_api(
path = '/account/verify_credentials.json',
parser = parse_user
)(self)
except TweepError, e:
raise TweepError('Failed to fetch username: %s' % e)
self._username = user.screen_name
return self.get_user(screen_name=self._username)
"""Show friends"""
friends = bind_api(
path = '/statuses/friends.json',
parser = parse_users,
allowed_param = ['id', 'user_id', 'screen_name', 'page']
)