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

WIP: General cursor (generator) like object for Twython functions #241

Merged
michaelhelmick merged 10 commits from dev/cursor into master 2013-07-20 12:01:29 -07:00
Conversation 1 Commits 10 Files changed 8 +181 -34
8 changed files with 181 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.travis.yml
View file

@ -29,7 +29,8 @@ env:
- PROTECTED_TWITTER_1=TwythonSecure1
- PROTECTED_TWITTER_2=TwythonSecure2
- TEST_TWEET_ID=332992304010899457
- TEST_LIST_ID=574
- TEST_LIST_SLUG=team
- TEST_LIST_OWNER_SCREEN_NAME=twitterapi
install: pip install -r requirements.txt
script: nosetests -v -w tests/ --logging-filter="twython" --with-cov --cov twython --cov-config .coveragerc --cov-report term-missing
notifications:

2
HISTORY.rst
View file

@ -10,6 +10,8 @@ History
- Pass ``client_args`` to the streaming ``__init__``, much like in core Twython (you can pass headers, timeout, hooks, proxies, etc.).
- Streamer has new parameter ``handlers`` which accepts a list of strings related to functions that are apart of the Streaming class and start with "on\_". i.e. ['delete'] is passed, when 'delete' is received from a stream response; ``on_delete`` will be called.
- When an actual request error happens and a ``RequestException`` is raised, it is caught and a ``TwythonError`` is raised instead for convenience.
- Added "cursor"-like functionality. Endpoints with the attribute ``iter_mode`` will be able to be passed to ``Twython.cursor`` and returned as a generator.
- ``Twython.search_gen`` has been deprecated. Please use ``twitter.cursor(twitter.search, q='your_query')`` instead, where ``twitter`` is your ``Twython`` instance.
3.0.0 (2013-06-18)
++++++++++++++++++

24
docs/index.rst
View file

@ -29,14 +29,38 @@ Features
Usage
-----
..
I know it isn't necessary to start a new tree for every section,
but I think it looks a bit cleaner that way!
.. toctree::
:maxdepth: 4
usage/install
.. toctree::
:maxdepth: 4
usage/starting_out
.. toctree::
:maxdepth: 4
usage/basic_usage
.. toctree::
:maxdepth: 4
usage/advanced_usage
.. toctree::
:maxdepth: 4
usage/streaming_api
.. toctree::
:maxdepth: 2
usage/special_functions
Twython API Documentation

43
docs/usage/special_functions.rst
View file

@ -7,6 +7,49 @@ This section covers methods to are part of Twython but not necessarily connected
*******************************************************************************
Cursor
------
This function returns a generator for Twitter API endpoints that are able to be pagintated in some way (either by cursor or since_id parameter)
The Old Way
^^^^^^^^^^^
.. code-block:: python
from twython import Twython
twitter = Twython(APP_KEY, APP_SECRET,
OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
results = twitter.search(q='twitter')
if results.get('statuses'):
for result in results['statuses']:
print result['id_str']
The New Way
^^^^^^^^^^^
.. code-block:: python
from twython import Twython
twitter = Twython(APP_KEY, APP_SECRET,
OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
results = twitter.cursor(t.search, q='twitter')
for result in results:
print result['id_str']
Another example:
.. code-block:: python
results = twitter.cursor(t.get_mentions_timeline)
for result in results:
print result['id_str']
HTML for Tweet
--------------

3
tests/config.py
View file

@ -16,7 +16,8 @@ protected_twitter_2 = os.environ.get('PROTECTED_TWITTER_2', 'TwythonSecure2')
# Test Ids
test_tweet_id = os.environ.get('TEST_TWEET_ID', '318577428610031617')
test_list_id = os.environ.get('TEST_LIST_ID', '574') # 574 is @twitter/team
test_list_slug = os.environ.get('TEST_LIST_SLUG', 'team')
test_list_owner_screen_name = os.environ.get('TEST_LIST_OWNER_SCREEN_NAME', 'twitterapi')
test_tweet_object = {u'contributors': None, u'truncated': False, u'text': u'http://t.co/FCmXyI6VHd is a #cool site, lol! @mikehelmick should #checkitout. If you can! #thanks Love, @__twython__ https://t.co/67pwRvY6z9', u'in_reply_to_status_id': None, u'id': 349683012054683648, u'favorite_count': 0, u'source': u'web', u'retweeted': False, u'coordinates': None, u'entities': {u'symbols': [], u'user_mentions': [{u'id': 29251354, u'indices': [45, 57], u'id_str': u'29251354', u'screen_name': u'mikehelmick', u'name': u'Mike Helmick'}, {u'id': 1431865928, u'indices': [104, 116], u'id_str': u'1431865928', u'screen_name': u'__twython__', u'name': u'Twython'}], u'hashtags': [{u'indices': [28, 33], u'text': u'cool'}, {u'indices': [65, 76], u'text': u'checkitout'}, {u'indices': [90, 97], u'text': u'thanks'}], u'urls': [{u'url': u'http://t.co/FCmXyI6VHd', u'indices': [0, 22], u'expanded_url': u'http://google.com', u'display_url': u'google.com'}, {u'url': u'https://t.co/67pwRvY6z9', u'indices': [117, 140], u'expanded_url': u'https://github.com', u'display_url': u'github.com'}]}, u'in_reply_to_screen_name': None, u'id_str': u'349683012054683648', u'retweet_count': 0, u'in_reply_to_user_id': None, u'favorited': False, u'user': {u'follow_request_sent': False, u'profile_use_background_image': True, u'default_profile_image': True, u'id': 1431865928, u'verified': False, u'profile_text_color': u'333333', u'profile_image_url_https': u'https://si0.twimg.com/sticky/default_profile_images/default_profile_3_normal.png', u'profile_sidebar_fill_color': u'DDEEF6', u'entities': {u'description': {u'urls': []}}, u'followers_count': 1, u'profile_sidebar_border_color': u'C0DEED', u'id_str': u'1431865928', u'profile_background_color': u'3D3D3D', u'listed_count': 0, u'profile_background_image_url_https': u'https://si0.twimg.com/images/themes/theme1/bg.png', u'utc_offset': None, u'statuses_count': 2, u'description': u'', u'friends_count': 1, u'location': u'', u'profile_link_color': u'0084B4', u'profile_image_url': u'http://a0.twimg.com/sticky/default_profile_images/default_profile_3_normal.png', u'following': False, u'geo_enabled': False, u'profile_background_image_url': u'http://a0.twimg.com/images/themes/theme1/bg.png', u'screen_name': u'__twython__', u'lang': u'en', u'profile_background_tile': False, u'favourites_count': 0, u'name': u'Twython', u'notifications': False, u'url': None, u'created_at': u'Thu May 16 01:11:09 +0000 2013', u'contributors_enabled': False, u'time_zone': None, u'protected': False, u'default_profile': False, u'is_translator': False}, u'geo': None, u'in_reply_to_user_id_str': None, u'possibly_sensitive': False, u'lang': u'en', u'created_at': u'Wed Jun 26 00:18:21 +0000 2013', u'in_reply_to_status_id_str': None, u'place': None}
test_tweet_html = '<a href="http://t.co/FCmXyI6VHd" class="twython-url">google.com</a> is a <a href="https://twitter.com/search?q=%23cool" class="twython-hashtag">#cool</a> site, lol! <a href="https://twitter.com/mikehelmick" class="twython-mention">@mikehelmick</a> should <a href="https://twitter.com/search?q=%23checkitout" class="twython-hashtag">#checkitout</a>. If you can! <a href="https://twitter.com/search?q=%23thanks" class="twython-hashtag">#thanks</a> Love, <a href="https://twitter.com/__twython__" class="twython-mention">@__twython__</a> <a href="https://t.co/67pwRvY6z9" class="twython-url">github.com</a>'

43
tests/test_core.py
View file

@ -3,8 +3,8 @@ from twython import Twython, TwythonError, TwythonAuthError
from .config import (
app_key, app_secret, oauth_token, oauth_token_secret,
protected_twitter_1, protected_twitter_2, screen_name,
test_tweet_id, test_list_id, access_token, test_tweet_object,
test_tweet_html
test_tweet_id, test_list_slug, test_list_owner_screen_name,
access_token, test_tweet_object, test_tweet_html
)
import time
@ -46,7 +46,7 @@ class TwythonAPITestCase(unittest.TestCase):
"""Test Twython generic POST request works, with a full url and
with just an endpoint"""
update_url = 'https://api.twitter.com/1.1/statuses/update.json'
status = self.api.post(update_url, params={'status': 'I love Twython!'})
status = self.api.post(update_url, params={'status': 'I love Twython! %s' % int(time.time())})
self.api.post('statuses/destroy/%s' % status['id_str'])
def test_get_lastfunction_header(self):
@ -65,9 +65,9 @@ class TwythonAPITestCase(unittest.TestCase):
self.assertRaises(TwythonError, self.api.get_lastfunction_header,
'no-api-call-was-made')
def test_search_gen(self):
def test_cursor(self):
"""Test looping through the generator results works, at least once that is"""
search = self.api.search_gen('twitter', count=1)
search = self.api.cursor(self.api.search, q='twitter', count=1)
counter = 0
while counter < 2:
counter += 1
@ -148,7 +148,7 @@ class TwythonAPITestCase(unittest.TestCase):
def test_update_and_destroy_status(self):
"""Test updating and deleting a status succeeds"""
status = self.api.update_status(status='Test post just to get deleted :(')
status = self.api.update_status(status='Test post just to get deleted :( %s' % int(time.time()))
self.api.destroy_status(id=status['id_str'])
def test_get_oembed_tweet(self):
@ -186,7 +186,7 @@ class TwythonAPITestCase(unittest.TestCase):
"""Test sending a direct message to someone who doesn't follow you
fails"""
self.assertRaises(TwythonError, self.api.send_direct_message,
screen_name=protected_twitter_2, text='Yo, man!')
screen_name=protected_twitter_2, text='Yo, man! %s' % int(time.time()))
# Friends & Followers
def test_get_user_ids_of_blocked_retweets(self):
@ -361,15 +361,16 @@ class TwythonAPITestCase(unittest.TestCase):
def test_get_list_statuses(self):
"""Test timeline of tweets authored by members of the
specified list succeeds"""
self.api.get_list_statuses(list_id=test_list_id)
self.api.get_list_statuses(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_create_update_destroy_list_add_remove_list_members(self):
"""Test create a list, adding and removing members then
deleting the list succeeds"""
the_list = self.api.create_list(name='Stuff')
the_list = self.api.create_list(name='Stuff %s' % int(time.time()))
list_id = the_list['id_str']
self.api.update_list(list_id=list_id, name='Stuff Renamed')
self.api.update_list(list_id=list_id, name='Stuff Renamed %s' % int(time.time()))
screen_names = ['johncena', 'xbox']
# Multi add/delete members
@ -386,28 +387,36 @@ class TwythonAPITestCase(unittest.TestCase):
def test_get_list_subscribers(self):
"""Test list of subscribers of a specific list succeeds"""
self.api.get_list_subscribers(list_id=test_list_id)
self.api.get_list_subscribers(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_subscribe_is_subbed_and_unsubscribe_to_list(self):
"""Test subscribing, is a list sub and unsubbing to list succeeds"""
self.api.subscribe_to_list(list_id=test_list_id)
self.api.subscribe_to_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
# Returns 404 if user is not a subscriber
self.api.is_list_subscriber(list_id=test_list_id,
self.api.is_list_subscriber(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name,
screen_name=screen_name)
self.api.unsubscribe_from_list(list_id=test_list_id)
self.api.unsubscribe_from_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_is_list_member(self):
"""Test returning if specified user is member of a list succeeds"""
# Returns 404 if not list member
self.api.is_list_member(list_id=test_list_id, screen_name='jack')
self.api.is_list_member(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name,
screen_name='themattharris')
def test_get_list_members(self):
"""Test listing members of the specified list succeeds"""
self.api.get_list_members(list_id=test_list_id)
self.api.get_list_members(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_get_specific_list(self):
"""Test getting specific list succeeds"""
self.api.get_specific_list(list_id=test_list_id)
self.api.get_specific_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_get_list_subscriptions(self):
"""Test collection of the lists the specified user is

60
twython/api.py
View file

@ -14,11 +14,16 @@ from requests.auth import HTTPBasicAuth
from requests_oauthlib import OAuth1, OAuth2
from . import __version__
from .advisory import TwythonDeprecationWarning
from .compat import json, urlencode, parse_qsl, quote_plus, str, is_py2
from .endpoints import EndpointsMixin
from .exceptions import TwythonError, TwythonAuthError, TwythonRateLimitError
from .helpers import _transparent_params
import warnings
warnings.simplefilter('always', TwythonDeprecationWarning) # For Python 2.7 >
class Twython(EndpointsMixin, object):
def __init__(self, app_key=None, app_secret=None, oauth_token=None,
@ -357,13 +362,19 @@ class Twython(EndpointsMixin, object):
)
return '%s?%s' % (api_url, '&'.join(querystring))
def search_gen(self, search_query, **params):
"""Returns a generator of tweets that match a specified query.
def search_gen(self, search_query, **params): # pragma: no cover
warnings.warn(
'This method is deprecated. You should use Twython.cursor instead. [eg. Twython.cursor(Twython.search, q=\'your_query\')]',
TwythonDeprecationWarning,
stacklevel=2
)
return self.cursor(self.search, q=search_query, **params)
Documentation: https://dev.twitter.com/docs/api/1.1/get/search/tweets
def cursor(self, function, **params):
"""Returns a generator for results that match a specified query.
:param search_query: Query you intend to search Twitter for
:param \*\*params: Extra parameters to send with your search request
:param function: Instance of a Twython function (Twython.get_home_timeline, Twython.search)
:param \*\*params: Extra parameters to send with your request (usually parameters excepted by the Twitter API endpoint)
:rtype: generator
Usage::
@ -371,27 +382,46 @@ class Twython(EndpointsMixin, object):
>>> from twython import Twython
>>> twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
>>> search = twitter.search_gen('python')
>>> for result in search:
>>> results = twitter.cursor(twitter.search, q='python')
>>> for result in results:
>>> print result
"""
content = self.search(q=search_query, **params)
if not hasattr(function, 'iter_mode'):
raise TwythonError('Unable to create generator for Twython method "%s"' % function.__name__)
if not content.get('statuses'):
content = function(**params)
if not content:
raise StopIteration
for tweet in content['statuses']:
yield tweet
if function.iter_mode == 'cursor' and content['next_cursor_str'] == '0':
raise StopIteration
if hasattr(function, 'iter_key'):
results = content.get(function.iter_key)
else:
results = content
for result in results:
yield result
try:
if not 'since_id' in params:
params['since_id'] = (int(content['statuses'][0]['id_str']) + 1)
if function.iter_mode == 'id':
if not 'max_id' in params:
# Add 1 to the id because since_id and max_id are inclusive
if hasattr(function, 'iter_metadata'):
since_id = content[function.iter_metadata].get('since_id_str')
else:
since_id = content[0]['id_str']
params['since_id'] = (int(since_id) - 1)
elif function.iter_mode == 'cursor':
params['cursor'] = content['next_cursor_str']
except (TypeError, ValueError): # pragma: no cover
raise TwythonError('Unable to generate next page of search results, `page` is not a number.')
for tweet in self.search_gen(search_query, **params):
yield tweet
for result in self.cursor(function, **params):
yield result
@staticmethod
def unicode2utf8(text):

37
twython/endpoints.py
View file

@ -24,6 +24,7 @@ class EndpointsMixin(object):
"""
return self.get('statuses/mentions_timeline', params=params)
get_mentions_timeline.iter_mode = 'id'
def get_user_timeline(self, **params):
"""Returns a collection of the most recent Tweets posted by the user
@ -33,6 +34,7 @@ class EndpointsMixin(object):
"""
return self.get('statuses/user_timeline', params=params)
get_user_timeline.iter_mode = 'id'
def get_home_timeline(self, **params):
"""Returns a collection of the most recent Tweets and retweets
@ -42,6 +44,7 @@ class EndpointsMixin(object):
"""
return self.get('statuses/home_timeline', params=params)
get_home_timeline.iter_mode = 'id'
def retweeted_of_me(self, **params):
"""Returns the most recent tweets authored by the authenticating user
@ -51,6 +54,7 @@ class EndpointsMixin(object):
"""
return self.get('statuses/retweets_of_me', params=params)
retweeted_of_me.iter_mode = 'id'
# Tweets
def get_retweets(self, **params):
@ -119,6 +123,8 @@ class EndpointsMixin(object):
"""
return self.get('statuses/retweeters/ids', params=params)
get_retweeters_ids.iter_mode = 'cursor'
get_retweeters_ids.iter_key = 'ids'
# Search
def search(self, **params):
@ -128,6 +134,9 @@ class EndpointsMixin(object):
"""
return self.get('search/tweets', params=params)
search.iter_mode = 'id'
search.iter_key = 'statuses'
search.iter_metadata = 'search_metadata'
# Direct Messages
def get_direct_messages(self, **params):
@ -137,6 +146,7 @@ class EndpointsMixin(object):
"""
return self.get('direct_messages', params=params)
get_direct_messages.iter_mode = 'id'
def get_sent_messages(self, **params):
"""Returns the 20 most recent direct messages sent by the authenticating user.
@ -145,6 +155,7 @@ class EndpointsMixin(object):
"""
return self.get('direct_messages/sent', params=params)
get_sent_messages.iter_mode = 'id'
def get_direct_message(self, **params):
"""Returns a single direct message, specified by an id parameter.
@ -188,6 +199,8 @@ class EndpointsMixin(object):
"""
return self.get('friends/ids', params=params)
get_friends_ids.iter_mode = 'cursor'
get_friends_ids.iter_key = 'ids'
def get_followers_ids(self, **params):
"""Returns a cursored collection of user IDs for every user
@ -197,6 +210,8 @@ class EndpointsMixin(object):
"""
return self.get('followers/ids', params=params)
get_followers_ids.iter_mode = 'cursor'
get_followers_ids.iter_key = 'ids'
def lookup_friendships(self, **params):
"""Returns the relationships of the authenticating user to the
@ -215,6 +230,8 @@ class EndpointsMixin(object):
"""
return self.get('friendships/incoming', params=params)
get_incoming_friendship_ids.iter_mode = 'cursor'
get_incoming_friendship_ids.iter_key = 'ids'
def get_outgoing_friendship_ids(self, **params):
"""Returns a collection of numeric IDs for every protected user for
@ -224,6 +241,8 @@ class EndpointsMixin(object):
"""
return self.get('friendships/outgoing', params=params)
get_outgoing_friendship_ids.iter_mode = 'cursor'
get_outgoing_friendship_ids.iter_key = 'ids'
def create_friendship(self, **params):
"""Allows the authenticating users to follow the user specified
@ -269,6 +288,8 @@ class EndpointsMixin(object):
"""
return self.get('friends/list', params=params)
get_friends_list.iter_mode = 'cursor'
get_friends_list.iter_key = 'users'
def get_followers_list(self, **params):
"""Returns a cursored collection of user objects for users
@ -278,6 +299,8 @@ class EndpointsMixin(object):
"""
return self.get('followers/list', params=params)
get_followers_list.iter_mode = 'cursor'
get_followers_list.iter_key = 'users'
# Users
def get_account_settings(self, **params):
@ -355,6 +378,8 @@ class EndpointsMixin(object):
"""
return self.get('blocks/list', params=params)
list_blocks.iter_mode = 'cursor'
list_blocks.iter_key = 'users'
def list_block_ids(self, **params):
"""Returns an array of numeric user ids the authenticating user is blocking.
@ -363,6 +388,8 @@ class EndpointsMixin(object):
"""
return self.get('blocks/ids', params=params)
list_block_ids.iter_mode = 'cursor'
list_block_ids.iter_key = 'ids'
def create_block(self, **params):
"""Blocks the specified user from following the authenticating user.
@ -481,6 +508,7 @@ class EndpointsMixin(object):
"""
return self.get('favorites/list', params=params)
get_favorites.iter_mode = 'id'
def destroy_favorite(self, **params):
"""Un-favorites the status specified in the ID parameter as the authenticating user.
@ -514,6 +542,7 @@ class EndpointsMixin(object):
"""
return self.get('lists/statuses', params=params)
get_list_statuses.iter_mode = 'id'
def delete_list_member(self, **params):
"""Removes the specified member from the list.
@ -530,6 +559,8 @@ class EndpointsMixin(object):
"""
return self.get('lists/subscribers', params=params)
get_list_subscribers.iter_mode = 'cursor'
get_list_subscribers.iter_key = 'users'
def subscribe_to_list(self, **params):
"""Subscribes the authenticated user to the specified list.
@ -579,6 +610,8 @@ class EndpointsMixin(object):
"""
return self.get('lists/members', params=params)
get_list_members.iter_mode = 'cursor'
get_list_members.iter_key = 'users'
def add_list_member(self, **params):
"""Add a member to a list.
@ -627,6 +660,8 @@ class EndpointsMixin(object):
"""
return self.get('lists/subscriptions', params=params)
get_list_subscriptions.iter_mode = 'cursor'
get_list_subscriptions.iter_key = 'lists'
def delete_list_members(self, **params):
"""Removes multiple members from a list, by specifying a
@ -644,6 +679,8 @@ class EndpointsMixin(object):
"""
return self.get('lists/ownerships', params=params)
show_owned_lists.iter_mode = 'cursor'
show_owned_lists.iter_key = 'lists'
# Saved Searches
def get_saved_searches(self, **params):