Commit c1886ad0 authored by Aaron Hill's avatar Aaron Hill
Browse files

Merge branch 'master' into add_update_banner

parents 579f2736 280f8d85
[run]
source = tweepy
......@@ -3,3 +3,6 @@
build/
dist/
tweepy.egg-info/
.env/
.coverage
htmlcov/
---
script: nosetests -v tests.test_api tests.test_streaming
language: python
python:
- "2.7"
install:
- pip install -r test_requirements.txt
script: ./run_tests.sh
env:
global:
- TWITTER_USERNAME="tweepytest"
......@@ -20,8 +24,5 @@ env:
nKkytraqLGUm33K1GpwkjOyxACDHYw4GMvOGyDwVTX7VNwqxbkUojB7qXYoQ
JjlEyFWS487IFteR87U9pt18qongJJIphaBdT9/lDVLsMWZ0Jh5ZLQfX+2jS
aF2UwsrYkzBUMrqMqYCc2+X6CuswLEZTVXDAlNh+emvhxZ5faMI=
python:
- "2.7"
branches:
only:
- master
after_success: 'coveralls'
Version 2.1
-----------
- Added get_oembed().
- friends() and followers() are back and updated to v1.1.
- Fixed report_spam() endpoint.
- Added "languages" parameter to streaming filter() method.
- Added "timeout" support for API object. Ex: API(timeout=1000).
- Python 2.5 no longer supported.
- Added compression support. Ex: API(compression=True).
- Added on_connect() callback to StreamListener.
- Switched API search() to v1.1 endpoint. Some breaking changes.
- Drop "page" based cursors and use "ID" based ones instead.
- [Compare
2.0...master](https://github.com/tweepy/tweepy/compare/2.0...master)
Version 2.0
-----------
_Dedicated in memory of Aaron Swartz_
......
import tweepy
# === Basic Authentication ===
#
# *Note: Basic Authentication is deprecated and no longer supported on Twitter.
# It is still provided for use in services like Status.net which still suppports it.*
#
# This mode of authentication requires the user provide their username and plain text password.
# These credentials will then be provided for each request to the API for authentication.
# You would normally fetch this in your application
# by asking the user or loading from some dark place.
username = ""
password = ""
# Create an authentication handler passing it the username and password.
# We will use this object later on when creating our API object.
auth = tweepy.auth.BasicAuthHandler(username, password)
# Create the API object providing it the authentication handler to use.
# Each request will then be authenticated using this handler.
api = tweepy.API(auth)
api.update_status('Updating using basic authentication via Tweepy!')
#! /usr/bin/env bash
if [[ $TRAVIS_SECURE_ENV_VARS == "false" ]]; then
USE_REPLAY=1 nosetests -v tests.test_api tests.test_utils
else
nosetests -v --with-coverage tests.test_api tests.test_streaming tests.test_cursors tests.test_utils
fi
......@@ -10,6 +10,6 @@ setup(name="tweepy",
author="Joshua Roesslein",
author_email="tweepy@googlegroups.com",
url="http://github.com/tweepy/tweepy",
packages = find_packages(),
keywords= "twitter library",
zip_safe = True)
packages=find_packages(exclude=['tests']),
keywords="twitter library",
zip_safe=True)
import os
from unittest import TestCase
from httreplay import start_replay, stop_replay
from httreplay.utils import filter_headers_key
from tweepy.auth import OAuthHandler
from tweepy.api import API
username = os.environ.get('TWITTER_USERNAME', '')
username = os.environ.get('TWITTER_USERNAME', 'tweepytest')
oauth_consumer_key = os.environ.get('CONSUMER_KEY', '')
oauth_consumer_secret = os.environ.get('CONSUMER_SECRET', '')
oauth_token = os.environ.get('ACCESS_KEY', '')
oauth_token_secret = os.environ.get('ACCESS_SECRET', '')
use_replay = os.environ.get('USE_REPLAY', False)
class TweepyTestCase(TestCase):
def setUp(self):
self.auth = create_auth()
self.api = API(self.auth)
self.api.retry_count = 2
self.api.retry_delay = 5
if use_replay:
start_replay('tests/record.json',
headers_key=filter_headers_key(['Authorization']))
def tearDown(self):
if use_replay:
stop_replay()
def create_auth():
auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)
......
This source diff could not be displayed because it is too large. You can view the blob instead.
......@@ -5,12 +5,11 @@ import os
from nose import SkipTest
from tweepy import (API, OAuthHandler, Friendship, Cursor,
MemoryCache, FileCache)
from config import *
from tweepy import Friendship, MemoryCache, FileCache
from config import TweepyTestCase, username, use_replay
test_tweet_id = '266367358078169089'
tweet_text = 'testing 1000'
"""Unit tests"""
......@@ -27,14 +26,7 @@ class TweepyErrorTests(unittest.TestCase):
self.assertEqual(e.reason, e2.reason)
self.assertEqual(e.response, e2.response)
class TweepyAPITests(unittest.TestCase):
def setUp(self):
auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)
auth.set_access_token(oauth_token, oauth_token_secret)
self.api = API(auth)
self.api.retry_count = 2
self.api.retry_delay = 5
class TweepyAPITests(TweepyTestCase):
# TODO: Actually have some sort of better assertion
def testgetoembed(self):
......@@ -62,12 +54,15 @@ class TweepyAPITests(unittest.TestCase):
def testretweets(self):
self.api.retweets(test_tweet_id)
def testretweeters(self):
self.api.retweeters(test_tweet_id)
def testgetstatus(self):
self.api.get_status(id=test_tweet_id)
def testupdateanddestroystatus(self):
# test update
text = 'testing %i' % random.randint(0, 1000)
text = tweet_text if use_replay else 'testing %i' % random.randint(0, 1000)
update = self.api.update_status(status=text)
self.assertEqual(update.text, text)
......@@ -82,6 +77,12 @@ class TweepyAPITests(unittest.TestCase):
u = self.api.get_user(783214)
self.assertEqual(u.screen_name, 'twitter')
def testlookupusers(self):
def check(users):
self.assertEqual(len(users), 2)
check(self.api.lookup_users(user_ids=[6844292, 6253282]))
check(self.api.lookup_users(screen_names=['twitterapi', 'twitter']))
def testsearchusers(self):
self.api.search_users('twitter')
......@@ -271,6 +272,7 @@ class TweepyAPITests(unittest.TestCase):
self.assertEqual(l.name, params['slug'])
assert_list(self.api.add_list_member(**params))
sleep(3)
assert_list(self.api.remove_list_member(**params))
def testlistmembers(self):
......@@ -291,7 +293,7 @@ class TweepyAPITests(unittest.TestCase):
self.api.list_subscribers('applepie', 'stars')
def testshowlistsubscriber(self):
self.assertTrue(self.api.show_list_subscriber('twitter', 'team', username))
self.assertTrue(self.api.show_list_subscriber('tweepytest', 'test', 'applepie'))
def testsavedsearches(self):
s = self.api.create_saved_search('test')
......@@ -315,43 +317,6 @@ class TweepyAPITests(unittest.TestCase):
self.assertTrue(place_name_in_list('Austin, TX',
self.api.reverse_geocode(lat=30.267370168467806, long= -97.74261474609375))) # Austin, TX, USA
class TweepyCursorTests(unittest.TestCase):
def setUp(self):
auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)
auth.set_access_token(oauth_token, oauth_token_secret)
self.api = API(auth)
self.api.retry_count = 2
self.api.retry_delay = 5
def testpagecursoritems(self):
items = list(Cursor(self.api.user_timeline).items())
self.assert_(len(items) > 0)
items = list(Cursor(self.api.user_timeline, 'twitter').items(30))
self.assert_(len(items) == 30)
def testpagecursorpages(self):
pages = list(Cursor(self.api.user_timeline).pages())
self.assert_(len(pages) > 0)
pages = list(Cursor(self.api.user_timeline, 'twitter').pages(5))
self.assert_(len(pages) == 5)
def testcursorcursoritems(self):
items = list(Cursor(self.api.friends_ids).items())
self.assert_(len(items) > 0)
items = list(Cursor(self.api.followers_ids, 'twitter').items(30))
self.assert_(len(items) == 30)
def testcursorcursorpages(self):
pages = list(Cursor(self.api.friends_ids).pages())
self.assert_(len(pages) > 0)
pages = list(Cursor(self.api.followers_ids, 'twitter').pages(5))
self.assert_(len(pages) == 5)
class TweepyCacheTests(unittest.TestCase):
timeout = 2.0
......
import unittest
from tweepy import API, Cursor
from config import create_auth
class TweepyCursorTests(unittest.TestCase):
def setUp(self):
self.api = API(create_auth())
def testidcursoritems(self):
items = list(Cursor(self.api.user_timeline).items(25))
self.assertEqual(len(items), 25)
def testidcursorpages(self):
pages = list(Cursor(self.api.user_timeline).pages(5))
self.assertEqual(len(pages), 5)
def testcursorcursoritems(self):
items = list(Cursor(self.api.friends_ids).items(10))
self.assertEqual(len(items), 10)
items = list(Cursor(self.api.followers_ids, 'twitter').items(10))
self.assertEqual(len(items), 10)
def testcursorcursorpages(self):
pages = list(Cursor(self.api.friends_ids).pages(1))
self.assert_(len(pages) == 1)
pages = list(Cursor(self.api.followers_ids, 'twitter').pages(1))
self.assert_(len(pages) == 1)
import unittest
from tweepy.models import ResultSet
class NoIdItem(object): pass
class IdItem(object):
def __init__(self, id):
self.id = id
ids_fixture = [1, 10, 8, 50, 2, 100, 5]
class TweepyResultSetTests(unittest.TestCase):
def setUp(self):
self.results = ResultSet()
for i in ids_fixture:
self.results.append(IdItem(i))
self.results.append(NoIdItem())
def testids(self):
ids = self.results.ids()
self.assertListEqual(ids, ids_fixture)
def testmaxid(self):
self.assertEqual(self.results.max_id, 100)
def testsinceid(self):
self.assertEqual(self.results.since_id, 1)
from unittest import TestCase
from tweepy.utils import *
class TweepyUtilsTests(TestCase):
def testparse_datetime(self):
result = parse_datetime("Wed Aug 27 13:08:45 +0000 2008")
self.assertEqual(datetime(2008, 8, 27, 13, 8, 45), result)
def testlist_to_csv(self):
self.assertEqual("1,2,3", list_to_csv([1,2,3]))
self.assertEqual("bird,tweet,nest,egg",
list_to_csv(["bird", "tweet", "nest", "egg"]))
......@@ -5,15 +5,15 @@
"""
Tweepy Twitter API library
"""
__version__ = '2.0'
__version__ = '2.1'
__author__ = 'Joshua Roesslein'
__license__ = 'MIT'
from tweepy.models import Status, User, DirectMessage, Friendship, SavedSearch, SearchResult, ModelFactory, Category
from tweepy.models import Status, User, DirectMessage, Friendship, SavedSearch, SearchResults, ModelFactory, Category
from tweepy.error import TweepError
from tweepy.api import API
from tweepy.cache import Cache, MemoryCache, FileCache
from tweepy.auth import BasicAuthHandler, OAuthHandler
from tweepy.auth import OAuthHandler
from tweepy.streaming import Stream, StreamListener
from tweepy.cursor import Cursor
......
......@@ -37,7 +37,7 @@ class API(object):
home_timeline = bind_api(
path = '/statuses/home_timeline.json',
payload_type = 'status', payload_list = True,
allowed_param = ['since_id', 'max_id', 'count', 'page'],
allowed_param = ['since_id', 'max_id', 'count'],
require_auth = True
)
......@@ -46,7 +46,7 @@ class API(object):
path = '/statuses/user_timeline.json',
payload_type = 'status', payload_list = True,
allowed_param = ['id', 'user_id', 'screen_name', 'since_id',
'max_id', 'count', 'page', 'include_rts']
'max_id', 'count', 'include_rts']
)
""" statuses/mentions """
......@@ -57,14 +57,6 @@ class API(object):
require_auth = True
)
"""/statuses/:id/retweeted_by.format"""
retweeted_by = bind_api(
path = '/statuses/{id}/retweeted_by.json',
payload_type = 'status', payload_list = True,
allowed_param = ['id', 'count', 'page'],
require_auth = True
)
"""/related_results/show/:id.format"""
related_results = bind_api(
path = '/related_results/show/{id}.json',
......@@ -73,19 +65,11 @@ class API(object):
require_auth = False
)
"""/statuses/:id/retweeted_by/ids.format"""
retweeted_by_ids = bind_api(
path = '/statuses/{id}/retweeted_by/ids.json',
payload_type = 'ids',
allowed_param = ['id', 'count', 'page'],
require_auth = True
)
""" statuses/retweets_of_me """
retweets_of_me = bind_api(
path = '/statuses/retweets_of_me.json',
payload_type = 'status', payload_list = True,
allowed_param = ['since_id', 'max_id', 'count', 'page'],
allowed_param = ['since_id', 'max_id', 'count'],
require_auth = True
)
......@@ -131,6 +115,12 @@ class API(object):
require_auth = True
)
retweeters = bind_api(
path = '/statuses/retweeters/ids.json',
payload_type = 'ids',
allowed_param = ['id', 'cursor', 'stringify_ids']
)
""" users/show """
get_user = bind_api(
path = '/users/show.json',
......@@ -195,7 +185,7 @@ class API(object):
direct_messages = bind_api(
path = '/direct_messages.json',
payload_type = 'direct_message', payload_list = True,
allowed_param = ['since_id', 'max_id', 'count', 'page'],
allowed_param = ['since_id', 'max_id', 'count'],
require_auth = True
)
......@@ -282,7 +272,7 @@ class API(object):
friends = bind_api(
path = '/friends/list.json',
payload_type = 'user', payload_list = True,
allowed_param = ['id', 'user_id', 'screen_name', 'page', 'cursor']
allowed_param = ['id', 'user_id', 'screen_name', 'cursor']
)
""" friendships/incoming """
......@@ -310,7 +300,7 @@ class API(object):
followers = bind_api(
path = '/followers/list.json',
payload_type = 'user', payload_list = True,
allowed_param = ['id', 'user_id', 'screen_name', 'page', 'cursor']
allowed_param = ['id', 'user_id', 'screen_name', 'cursor']
)
""" account/verify_credentials """
......@@ -443,7 +433,7 @@ class API(object):
blocks = bind_api(
path = '/blocks/list.json',
payload_type = 'user', payload_list = True,
allowed_param = ['page'],
allowed_param = ['cursor'],
require_auth = True
)
......@@ -639,12 +629,10 @@ class API(object):
""" search """
search = bind_api(
search_api = True,
path = '/search.json',
payload_type = 'search_result', payload_list = True,
allowed_param = ['q', 'lang', 'locale', 'rpp', 'page', 'since_id', 'geocode', 'show_user', 'max_id', 'since', 'until', 'result_type']
path = '/search/tweets.json',
payload_type = 'search_results',
allowed_param = ['q', 'lang', 'locale', 'since_id', 'geocode', 'show_user', 'max_id', 'since', 'until', 'result_type']
)
search.pagination_mode = 'page'
""" trends/daily """
trends_daily = bind_api(
......
......@@ -21,19 +21,6 @@ class AuthHandler(object):
raise NotImplementedError
class BasicAuthHandler(AuthHandler):
def __init__(self, username, password):
self.username = username
self._b64up = base64.b64encode('%s:%s' % (username, password))
def apply_auth(self, url, method, headers, parameters):
headers['Authorization'] = 'Basic %s' % self._b64up
def get_username(self):
return self.username
class OAuthHandler(AuthHandler):
"""OAuth authentication handler"""
......
......@@ -200,6 +200,9 @@ def bind_api(**config):
# Set pagination mode
if 'cursor' in APIMethod.allowed_param:
_call.pagination_mode = 'cursor'
elif 'max_id' in APIMethod.allowed_param and \
'since_id' in APIMethod.allowed_param:
_call.pagination_mode = 'id'
elif 'page' in APIMethod.allowed_param:
_call.pagination_mode = 'page'
......
......@@ -11,8 +11,12 @@ class Cursor(object):
if hasattr(method, 'pagination_mode'):
if method.pagination_mode == 'cursor':
self.iterator = CursorIterator(method, args, kargs)
else:
elif method.pagination_mode == 'id':
self.iterator = IdIterator(method, args, kargs)
elif method.pagination_mode == 'page':
self.iterator = PageIterator(method, args, kargs)
else:
raise TweepError('Invalid pagination mode.')
else:
raise TweepError('This method does not perform pagination')
......@@ -74,6 +78,44 @@ class CursorIterator(BaseIterator):
self.count -= 1
return data
class IdIterator(BaseIterator):
def __init__(self, method, args, kargs):
BaseIterator.__init__(self, method, args, kargs)
self.max_id = kargs.get('max_id')
self.since_id = kargs.get('since_id')
self.count = 0
def next(self):
"""Fetch a set of items with IDs less than current set."""
if self.limit and self.limit == self.count:
raise StopIteration
# max_id is inclusive so decrement by one
# to avoid requesting duplicate items.
max_id = self.since_id - 1 if self.max_id else None
data = self.method(max_id = max_id, *self.args, **self.kargs)
if len(data) == 0:
raise StopIteration
self.max_id = data.max_id
self.since_id = data.since_id
self.count += 1
return data
def prev(self):
"""Fetch a set of items with IDs greater than current set."""
if self.limit and self.limit == self.count:
raise StopIteration
since_id = self.max_id
data = self.method(since_id = since_id, *self.args, **self.kargs)
if len(data) == 0:
raise StopIteration
self.max_id = data.max_id
self.since_id = data.since_id
self.count += 1
return data
class PageIterator(BaseIterator):
def __init__(self, method, args, kargs):
......
......@@ -3,13 +3,32 @@
# See LICENSE for details.
from tweepy.error import TweepError
from tweepy.utils import parse_datetime, parse_html_value, parse_a_href, \
parse_search_datetime, unescape_html
from tweepy.utils import parse_datetime, parse_html_value, parse_a_href
class ResultSet(list):
"""A list like object that holds results from a Twitter API query."""
def __init__(self, max_id=None, since_id=None):
super(ResultSet, self).__init__()
self._max_id = max_id
self._since_id = since_id
@property
def max_id(self):
if self._max_id:
return self._max_id
ids = self.ids()
return max(ids) if ids else None
@property
def since_id(self):
if self._since_id:
return self._since_id
ids = self.ids()
return min(ids) if ids else None
def ids(self):
return [item.id for item in self if hasattr(item, 'id')]
class Model(object):
......@@ -209,34 +228,18 @@ class SavedSearch(Model):
return self._api.destroy_saved_search(self.id)
class SearchResult(Model):
class SearchResults(ResultSet):
@classmethod
def parse(cls, api, json):
result = cls()
for k, v in json.items():
if k == 'created_at':
setattr(result, k, parse_search_datetime(v))
elif k == 'source':
setattr(result, k, parse_html_value(unescape_html(v)))
else:
setattr(