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

Merge branch 'master' of github:ryanmcgrath/twython

Browse source 
Conflicts:
	examples/block_spammer.py
	examples/block_user.py
	examples/unblock_user.py
	examples/unfollow_user.py
This commit is contained in:
Ben McGinnes 2014-08-06 01:05:01 +10:00
commit 5300373299
17 changed files with 319 additions and 235 deletions
Download patch file Download diff file Expand all files Collapse all files

20
examples/block_spammer.py
View file

@ -1,20 +0,0 @@
from twython import Twython, TwythonError
# Optionally accept user data from the command line (or elsewhere).
#
# Usage: block_spammer.py SomeoneAnnoying
import sys
if len(sys.argv) >= 2:
target = sys.argv[1]
else:
target = raw_input("User to follow: ") # For Python 3.x use: target = input("User to follow: ")
# Requires Authentication as of Twitter API v1.1
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
try:
twitter.report_spam(screen_name=target)
except TwythonError as e:
print(e)

20
examples/block_user.py
View file

@ -1,20 +0,0 @@
from twython import Twython, TwythonError
# Optionally accept user data from the command line (or elsewhere).
#
# Usage: block_user.py A_Twitter_Troll
import sys
if len(sys.argv) >= 2:
target = sys.argv[1]
else:
target = raw_input("User to follow: ") # For Python 3.x use: target = input("User to follow: ")
# Requires Authentication as of Twitter API v1.1
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
try:
twitter.create_block(screen_name=target, skip_status="true")
except TwythonError as e:
print(e)

3
examples/follow_user.py
View file

@ -9,7 +9,8 @@ import sys
if len(sys.argv) >= 2:
target = sys.argv[1]
else:
target = raw_input("User to follow: ") # For Python 3.x use: target = input("User to follow: ")
target = raw_input("User to follow: ")
# For Python 3.x use: target = input("User to follow: ")
# Requires Authentication as of Twitter API v1.1
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)

4
examples/search_results.py
View file

@ -8,5 +8,7 @@ except TwythonError as e:
print e
for tweet in search_results['statuses']:
print 'Tweet from @%s Date: %s' % (tweet['user']['screen_name'].encode('utf-8'), tweet['created_at'])
print 'Tweet from @%s Date: %s' % (tweet['user']['screen_nam\
e'].encode('utf-8'),
tweet['created_at'])
print tweet['text'].encode('utf-8'), '\n'

6
examples/stream.py
View file

@ -16,5 +16,7 @@ stream = MyStreamer(APP_KEY, APP_SECRET,
OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
stream.statuses.filter(track='twitter')
#stream.user() # Read the authenticated users home timeline (what they see on Twitter) in real-time
#stream.site(follow='twitter')
# stream.user()
# Read the authenticated users home timeline
# (what they see on Twitter) in real-time
# stream.site(follow='twitter')

20
examples/unblock_user.py
View file

@ -1,20 +0,0 @@
from twython import Twython, TwythonError
# Optionally accept user data from the command line (or elsewhere).
#
# Usage: unblock_user.py Not_So_Bad_After_All
import sys
if len(sys.argv) >= 2:
target = sys.argv[1]
else:
target = raw_input("User to follow: ") # For Python 3.x use: target = input("User to follow: ")
# Requires Authentication as of Twitter API v1.1
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
try:
twitter.destroy_block(screen_name=target, skip_status="true")
except TwythonError as e:
print(e)

20
examples/unfollow_user.py
View file

@ -1,20 +0,0 @@
from twython import Twython, TwythonError
# Optionally accept user data from the command line (or elsewhere).
#
# Usage: unfollow_user.py Not_So_Cool_Really
import sys
if len(sys.argv) >= 2:
target = sys.argv[1]
else:
target = raw_input("User to follow: ") # For Python 3.x use: target = input("User to follow: ")
# Requires Authentication as of Twitter API v1.1
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
try:
twitter.destroy_friendship(screen_name=target)
except TwythonError as e:
print(e)

5
setup.py
View file

@ -29,9 +29,10 @@ setup(
license=open('LICENSE').read(),
url='https://github.com/ryanmcgrath/twython/tree/master',
keywords='twitter search api tweet twython stream',
description='Actively maintained, pure Python wrapper for the Twitter API. Supports both normal and streaming Twitter APIs',
description='Actively maintained, pure Python wrapper for the \
Twitter API. Supports both normal and streaming Twitter APIs',
long_description=open('README.rst').read() + '\n\n' +
open('HISTORY.rst').read(),
open('HISTORY.rst').read(),
include_package_data=True,
packages=packages,
classifiers=[

3
tests/config.py
View file

@ -23,7 +23,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_slug = os.environ.get('TEST_LIST_SLUG', 'team')
test_list_owner_screen_name = os.environ.get('TEST_LIST_OWNER_SCREEN_NAME', 'twitterapi')
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>'

12
tests/test_core.py
View file

@ -28,8 +28,8 @@ class TwythonAPITestCase(unittest.TestCase):
return '%s/%s.json' % (self.api.api_url % self.api.api_version, endpoint)
def register_response(self, method, url, body='{}', match_querystring=False,
status=200, adding_headers=None, stream=False,
content_type='application/json; charset=utf-8'):
status=200, adding_headers=None, stream=False,
content_type='application/json; charset=utf-8'):
"""Wrapper function for responses for simpler unit tests"""
# responses uses BytesIO to hold the body so it needs to be in bytes
@ -37,7 +37,7 @@ class TwythonAPITestCase(unittest.TestCase):
body = bytes(body, 'UTF-8')
responses.add(method, url, body, match_querystring,
status, adding_headers, stream, content_type)
status, adding_headers, stream, content_type)
@responses.activate
def test_request_should_handle_full_endpoint(self):
@ -77,7 +77,7 @@ class TwythonAPITestCase(unittest.TestCase):
@responses.activate
def test_request_should_throw_exception_with_invalid_http_method(self):
"""Test that request() throws an exception when an invalid HTTP method is passed"""
#TODO(cash): should Twython catch the AttributeError and throw a TwythonError
# TODO(cash): should Twython catch the AttributeError and throw a TwythonError
self.assertRaises(AttributeError, self.api.request, endpoint='search/tweets', method='INVALID')
@responses.activate
@ -315,5 +315,5 @@ class TwythonAPITestCase(unittest.TestCase):
"""Test using expanded url in HTML for Tweet displays full urls"""
tweet_text = self.api.html_for_tweet(test_tweet_object, False)
# Make sure HTML doesn't contain the display OR expanded url
self.assertTrue(not 'http://google.com' in tweet_text)
self.assertTrue(not 'google.com' in tweet_text)
self.assertTrue('http://google.com' not in tweet_text)
self.assertTrue('google.com' not in tweet_text)

22
tests/test_endpoints.py
View file

@ -20,8 +20,10 @@ class TwythonEndpointsTestCase(unittest.TestCase):
'allow_redirects': False
}
# This is so we can hit coverage that Twython sets
# User-Agent for us if none is supplied
oauth2_client_args = {
'headers': {} # This is so we can hit coverage that Twython sets User-Agent for us if none is supplied
'headers': {}
}
self.api = Twython(app_key, app_secret,
@ -42,7 +44,8 @@ class TwythonEndpointsTestCase(unittest.TestCase):
"""Test returning timeline for authenticated user and random user
succeeds"""
self.api.get_user_timeline() # Authenticated User Timeline
self.api.get_user_timeline(screen_name='twitter') # Random User Timeline
self.api.get_user_timeline(screen_name='twitter')
# Random User Timeline
@unittest.skip('skipping non-updated test')
def test_get_protected_user_timeline_following(self):
@ -82,7 +85,8 @@ class TwythonEndpointsTestCase(unittest.TestCase):
@unittest.skip('skipping non-updated test')
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 :( %s' % int(time.time()))
status = self.api.update_status(status='Test post just to get \
deleted :( %s' % int(time.time()))
self.api.destroy_status(id=status['id_str'])
@unittest.skip('skipping non-updated test')
@ -117,7 +121,8 @@ class TwythonEndpointsTestCase(unittest.TestCase):
def test_send_get_and_destroy_direct_message(self):
"""Test sending, getting, then destory a direct message succeeds"""
message = self.api.send_direct_message(screen_name=protected_twitter_1,
text='Hey d00d! %s' % int(time.time()))
text='Hey d00d! %s\
' % int(time.time()))
self.api.get_direct_message(id=message['id_str'])
self.api.destroy_direct_message(id=message['id_str'])
@ -127,7 +132,8 @@ class TwythonEndpointsTestCase(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! %s' % int(time.time()))
screen_name=protected_twitter_2, text='Yo, man! \
%s' % int(time.time()))
# Friends & Followers
@unittest.skip('skipping non-updated test')
@ -348,7 +354,8 @@ class TwythonEndpointsTestCase(unittest.TestCase):
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 %s' % int(time.time()))
self.api.update_list(list_id=list_id, name='Stuff Renamed \
%s' % int(time.time()))
screen_names = ['johncena', 'xbox']
# Multi add/delete members
@ -359,7 +366,8 @@ class TwythonEndpointsTestCase(unittest.TestCase):
# Single add/delete member
self.api.add_list_member(list_id=list_id, screen_name='justinbieber')
self.api.delete_list_member(list_id=list_id, screen_name='justinbieber')
self.api.delete_list_member(list_id=list_id,
screen_name='justinbieber')
self.api.delete_list(list_id=list_id)

5
twython/__init__.py
View file

@ -11,8 +11,9 @@ Twython
Twython is a library for Python that wraps the Twitter API.
It aims to abstract away all the API endpoints, so that additions to the library
and/or the Twitter API won't cause any overall problems.
It aims to abstract away all the API endpoints, so that
additions to the library and/or the Twitter API won't
cause any overall problems.
Questions, comments? ryan@venodesigns.net
"""

5
twython/advisory.py
View file

@ -15,7 +15,8 @@ only TwythonDeprecationWarnings.
class TwythonDeprecationWarning(DeprecationWarning):
"""Custom DeprecationWarning to be raised when methods/variables are being deprecated in Twython.
Python 2.7 > ignores DeprecationWarning so we want to specifcally bubble up ONLY Twython Deprecation Warnings
"""Custom DeprecationWarning to be raised when methods/variables
are being deprecated in Twython. Python 2.7 > ignores DeprecationWarning
so we want to specifcally bubble up ONLY Twython Deprecation Warnings
"""
pass

212
twython/api.py
View file

@ -27,29 +27,41 @@ warnings.simplefilter('always', TwythonDeprecationWarning) # For Python 2.7 >
class Twython(EndpointsMixin, object):
def __init__(self, app_key=None, app_secret=None, oauth_token=None,
oauth_token_secret=None, access_token=None, token_type='bearer',
oauth_version=1, api_version='1.1', client_args=None, auth_endpoint='authenticate'):
"""Instantiates an instance of Twython. Takes optional parameters for authentication and such (see below).
oauth_token_secret=None, access_token=None,
token_type='bearer', oauth_version=1, api_version='1.1',
client_args=None, auth_endpoint='authenticate'):
"""Instantiates an instance of Twython. Takes optional parameters for
authentication and such (see below).
:param app_key: (optional) Your applications key
:param app_secret: (optional) Your applications secret key
:param oauth_token: (optional) When using **OAuth 1**, combined with oauth_token_secret to make authenticated calls
:param oauth_token_secret: (optional) When using **OAuth 1** combined with oauth_token to make authenticated calls
:param access_token: (optional) When using **OAuth 2**, provide a valid access token if you have one
:param token_type: (optional) When using **OAuth 2**, provide your token type. Default: bearer
:param oauth_version: (optional) Choose which OAuth version to use. Default: 1
:param api_version: (optional) Choose which Twitter API version to use. Default: 1.1
:param client_args: (optional) Accepts some requests Session parameters and some requests Request parameters.
See http://docs.python-requests.org/en/latest/api/#sessionapi and requests section below it for details.
[ex. headers, proxies, verify(SSL verification)]
:param auth_endpoint: (optional) Lets you select which authentication endpoint will use your application.
This will allow the application to have DM access if the endpoint is 'authorize'.
Default: authenticate.
:param oauth_token: (optional) When using **OAuth 1**, combined with
oauth_token_secret to make authenticated calls
:param oauth_token_secret: (optional) When using **OAuth 1** combined
with oauth_token to make authenticated calls
:param access_token: (optional) When using **OAuth 2**, provide a
valid access token if you have one
:param token_type: (optional) When using **OAuth 2**, provide your
token type. Default: bearer
:param oauth_version: (optional) Choose which OAuth version to use.
Default: 1
:param api_version: (optional) Choose which Twitter API version to
use. Default: 1.1
:param client_args: (optional) Accepts some requests Session parameters
and some requests Request parameters.
See http://docs.python-requests.org/en/latest/api/#sessionapi
and requests section below it for details.
[ex. headers, proxies, verify(SSL verification)]
:param auth_endpoint: (optional) Lets you select which authentication
endpoint will use your application.
This will allow the application to have DM access
if the endpoint is 'authorize'.
Default: authenticate.
"""
# API urls, OAuth urls and API version; needed for hitting that there API.
# API urls, OAuth urls and API version; needed for hitting that there
# API.
self.api_version = api_version
self.api_url = 'https://api.twitter.com/%s'
@ -75,16 +87,18 @@ class Twython(EndpointsMixin, object):
self.client_args = client_args or {}
default_headers = {'User-Agent': 'Twython v' + __version__}
if not 'headers' in self.client_args:
if 'headers' not in self.client_args:
# If they didn't set any headers, set our defaults for them
self.client_args['headers'] = default_headers
elif 'User-Agent' not in self.client_args['headers']:
# If they set headers, but didn't include User-Agent.. set it for them
# If they set headers, but didn't include User-Agent.. set
# it for them
self.client_args['headers'].update(default_headers)
# Generate OAuth authentication object for the request
# If no keys/tokens are passed to __init__, auth=None allows for
# unauthenticated requests, although I think all v1.1 requests need auth
# unauthenticated requests, although I think all v1.1 requests
# need auth
auth = None
if oauth_version == 1:
# User Authentication is through OAuth 1
@ -93,12 +107,14 @@ class Twython(EndpointsMixin, object):
auth = OAuth1(self.app_key, self.app_secret)
if self.app_key is not None and self.app_secret is not None and \
self.oauth_token is not None and self.oauth_token_secret is not None:
self.oauth_token is not None and self.oauth_token_secret is \
not None:
auth = OAuth1(self.app_key, self.app_secret,
self.oauth_token, self.oauth_token_secret)
elif oauth_version == 2 and self.access_token:
# Application Authentication is through OAuth 2
token = {'token_type': token_type, 'access_token': self.access_token}
token = {'token_type': token_type,
'access_token': self.access_token}
auth = OAuth2(self.app_key, token=token)
self.client = requests.Session()
@ -166,21 +182,26 @@ class Twython(EndpointsMixin, object):
ExceptionType = TwythonError
if response.status_code == 429:
# Twitter API 1.1, always return 429 when rate limit is exceeded
# Twitter API 1.1, always return 429 when
# rate limit is exceeded
ExceptionType = TwythonRateLimitError
elif response.status_code == 401 or 'Bad Authentication data' in error_message:
elif response.status_code == 401 or 'Bad Authentication data' \
in error_message:
# Twitter API 1.1, returns a 401 Unauthorized or
# a 400 "Bad Authentication data" for invalid/expired app keys/user tokens
# a 400 "Bad Authentication data" for invalid/expired
# app keys/user tokens
ExceptionType = TwythonAuthError
raise ExceptionType(error_message,
error_code=response.status_code,
retry_after=response.headers.get('retry-after'))
retry_after=response.headers.get('retry-\
after'))
try:
content = response.json()
except ValueError:
raise TwythonError('Response was not valid JSON. Unable to decode.')
raise TwythonError('Response was not valid JSON. \
Unable to decode.')
return content
@ -190,7 +211,8 @@ class Twython(EndpointsMixin, object):
error_message = 'An error occurred processing your request.'
try:
content = response.json()
# {"errors":[{"code":34,"message":"Sorry, that page does not exist"}]}
# {"errors":[{"code":34,"message":"Sorry,
# that page does not exist"}]}
error_message = content['errors'][0]['message']
except ValueError:
# bad json data from Twitter for an error
@ -204,13 +226,18 @@ class Twython(EndpointsMixin, object):
def request(self, endpoint, method='GET', params=None, version='1.1'):
"""Return dict of response received from Twitter's API
:param endpoint: (required) Full url or Twitter API endpoint (e.g. search/tweets)
:param endpoint: (required) Full url or Twitter API endpoint
(e.g. search/tweets)
:type endpoint: string
:param method: (optional) Method of accessing data, either GET or POST. (default GET)
:param method: (optional) Method of accessing data, either
GET or POST. (default GET)
:type method: string
:param params: (optional) Dict of parameters (if any) accepted the by Twitter API endpoint you are trying to access (default None)
:param params: (optional) Dict of parameters (if any) accepted
the by Twitter API endpoint you are trying to
access (default None)
:type params: dict or None
:param version: (optional) Twitter API version to access (default 1.1)
:param version: (optional) Twitter API version to access
(default 1.1)
:type version: string
:rtype: dict
@ -223,7 +250,8 @@ class Twython(EndpointsMixin, object):
else:
url = '%s/%s.json' % (self.api_url % version, endpoint)
content = self._request(url, method=method, params=params, api_call=url)
content = self._request(url, method=method, params=params,
api_call=url)
return content
@ -239,7 +267,8 @@ class Twython(EndpointsMixin, object):
"""Returns a specific header from the last API call
This will return None if the header is not present
:param header: (required) The name of the header you want to get the value of
:param header: (required) The name of the header you want to get
the value of
Most useful for the following header information:
x-rate-limit-limit,
@ -249,21 +278,31 @@ class Twython(EndpointsMixin, object):
"""
if self._last_call is None:
raise TwythonError('This function must be called after an API call. It delivers header information.')
raise TwythonError('This function must be called after an API call. \
It delivers header information.')
return self._last_call['headers'].get(header, default_return_value)
def get_authentication_tokens(self, callback_url=None, force_login=False, screen_name=''):
"""Returns a dict including an authorization URL, ``auth_url``, to direct a user to
def get_authentication_tokens(self, callback_url=None, force_login=False,
screen_name=''):
"""Returns a dict including an authorization URL, ``auth_url``, to
direct a user to
:param callback_url: (optional) Url the user is returned to after they authorize your app (web clients only)
:param force_login: (optional) Forces the user to enter their credentials to ensure the correct users account is authorized.
:param screen_name: (optional) If forced_login is set OR user is not currently logged in, Prefills the username input box of the OAuth login screen with the given value
:param callback_url: (optional) Url the user is returned to after
they authorize your app (web clients only)
:param force_login: (optional) Forces the user to enter their
credentials to ensure the correct users
account is authorized.
:param screen_name: (optional) If forced_login is set OR user is
not currently logged in, Prefills the username
input box of the OAuth login screen with the
given value
:rtype: dict
"""
if self.oauth_version != 1:
raise TwythonError('This method can only be called when your OAuth version is 1.0.')
raise TwythonError('This method can only be called when your \
OAuth version is 1.0.')
request_args = {}
if callback_url:
@ -271,15 +310,18 @@ class Twython(EndpointsMixin, object):
response = self.client.get(self.request_token_url, params=request_args)
if response.status_code == 401:
raise TwythonAuthError(response.content, error_code=response.status_code)
raise TwythonAuthError(response.content,
error_code=response.status_code)
elif response.status_code != 200:
raise TwythonError(response.content, error_code=response.status_code)
raise TwythonError(response.content,
error_code=response.status_code)
request_tokens = dict(parse_qsl(response.content.decode('utf-8')))
if not request_tokens:
raise TwythonError('Unable to decode request tokens.')
oauth_callback_confirmed = request_tokens.get('oauth_callback_confirmed') == 'true'
oauth_callback_confirmed = request_tokens.get('oauth_callback_confirmed') \
== 'true'
auth_url_params = {
'oauth_token': request_tokens['oauth_token'],
@ -295,21 +337,28 @@ class Twython(EndpointsMixin, object):
if callback_url and not oauth_callback_confirmed:
auth_url_params['oauth_callback'] = self.callback_url
request_tokens['auth_url'] = self.authenticate_url + '?' + urlencode(auth_url_params)
request_tokens['auth_url'] = self.authenticate_url + \
'?' + urlencode(auth_url_params)
return request_tokens
def get_authorized_tokens(self, oauth_verifier):
"""Returns a dict of authorized tokens after they go through the :class:`get_authentication_tokens` phase.
"""Returns a dict of authorized tokens after they go through the
:class:`get_authentication_tokens` phase.
:param oauth_verifier: (required) The oauth_verifier (or a.k.a PIN for non web apps) retrieved from the callback url querystring
:param oauth_verifier: (required) The oauth_verifier (or a.k.a PIN
for non web apps) retrieved from the callback url querystring
:rtype: dict
"""
if self.oauth_version != 1:
raise TwythonError('This method can only be called when your OAuth version is 1.0.')
raise TwythonError('This method can only be called when your \
OAuth version is 1.0.')
response = self.client.get(self.access_token_url, params={'oauth_verifier': oauth_verifier}, headers={'Content-Type': 'application/json'})
response = self.client.get(self.access_token_url,
params={'oauth_verifier': oauth_verifier},
headers={'Content-Type': 'application/\
json'})
if response.status_code == 401:
try:
@ -322,7 +371,8 @@ class Twython(EndpointsMixin, object):
except ValueError:
content = {}
raise TwythonError(content.get('error', 'Invalid / expired Token'), error_code=response.status_code)
raise TwythonError(content.get('error', 'Invalid / expired To \
ken'), error_code=response.status_code)
authorized_tokens = dict(parse_qsl(response.content.decode('utf-8')))
if not authorized_tokens:
@ -331,12 +381,14 @@ class Twython(EndpointsMixin, object):
return authorized_tokens # pragma: no cover
def obtain_access_token(self):
"""Returns an OAuth 2 access token to make OAuth 2 authenticated read-only calls.
"""Returns an OAuth 2 access token to make OAuth 2 authenticated
read-only calls.
:rtype: string
"""
if self.oauth_version != 2:
raise TwythonError('This method can only be called when your OAuth version is 2.0.')
raise TwythonError('This method can only be called when your \
OAuth version is 2.0.')
data = {'grant_type': 'client_credentials'}
basic_auth = HTTPBasicAuth(self.app_key, self.app_secret)
@ -358,8 +410,10 @@ class Twython(EndpointsMixin, object):
def construct_api_url(api_url, **params):
"""Construct a Twitter API url, encoded, with parameters
:param api_url: URL of the Twitter API endpoint you are attempting to construct
:param \*\*params: Parameters that are accepted by Twitter for the endpoint you're requesting
:param api_url: URL of the Twitter API endpoint you are attempting
to construct
:param \*\*params: Parameters that are accepted by Twitter for the
endpoint you're requesting
:rtype: string
Usage::
@ -368,7 +422,8 @@ class Twython(EndpointsMixin, object):
>>> twitter = Twython()
>>> api_url = 'https://api.twitter.com/1.1/search/tweets.json'
>>> constructed_url = twitter.construct_api_url(api_url, q='python', result_type='popular')
>>> constructed_url = twitter.construct_api_url(api_url, q='python',
result_type='popular')
>>> print constructed_url
https://api.twitter.com/1.1/search/tweets.json?q=python&result_type=popular
@ -384,7 +439,8 @@ class Twython(EndpointsMixin, object):
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\')]',
'This method is deprecated. You should use Twython.cursor instead. \
[eg. Twython.cursor(Twython.search, q=\'your_query\')]',
TwythonDeprecationWarning,
stacklevel=2
)
@ -393,14 +449,17 @@ class Twython(EndpointsMixin, object):
def cursor(self, function, return_pages=False, **params):
"""Returns a generator for results that match a specified query.
:param function: Instance of a Twython function (Twython.get_home_timeline, Twython.search)
:param \*\*params: Extra parameters to send with your request (usually parameters accepted by the Twitter API endpoint)
:param function: Instance of a Twython function
(Twython.get_home_timeline, Twython.search)
:param \*\*params: Extra parameters to send with your request
(usually parameters accepted by the Twitter API endpoint)
:rtype: generator
Usage::
>>> from twython import Twython
>>> twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
>>> twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN,
OAUTH_TOKEN_SECRET)
>>> results = twitter.cursor(twitter.search, q='python')
>>> for result in results:
@ -408,7 +467,8 @@ class Twython(EndpointsMixin, object):
"""
if not hasattr(function, 'iter_mode'):
raise TwythonError('Unable to create generator for Twython method "%s"' % function.__name__)
raise TwythonError('Unable to create generator for Twython \
method "%s"' % function.__name__)
while True:
content = function(**params)
@ -427,22 +487,26 @@ class Twython(EndpointsMixin, object):
for result in results:
yield result
if function.iter_mode == 'cursor' and content['next_cursor_str'] == '0':
if function.iter_mode == 'cursor' and \
content['next_cursor_str'] == '0':
raise StopIteration
try:
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 'max_id' not 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')
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.')
raise TwythonError('Unable to generate next page of search \
results, `page` is not a number.')
@staticmethod
def unicode2utf8(text):
@ -464,11 +528,14 @@ class Twython(EndpointsMixin, object):
"""Return HTML for a tweet (urls, mentions, hashtags replaced with links)
:param tweet: Tweet object from received from Twitter API
:param use_display_url: Use display URL to represent link (ex. google.com, github.com). Default: True
:param use_expanded_url: Use expanded URL to represent link (e.g. http://google.com). Default False
:param use_display_url: Use display URL to represent link
(ex. google.com, github.com). Default: True
:param use_expanded_url: Use expanded URL to represent link
(e.g. http://google.com). Default False
If use_expanded_url is True, it overrides use_display_url.
If use_display_url and use_expanded_url is False, short url will be used (t.co/xxxxx)
If use_display_url and use_expanded_url is False, short url will
be used (t.co/xxxxx)
"""
if 'retweeted_status' in tweet:
@ -483,7 +550,8 @@ class Twython(EndpointsMixin, object):
start, end = entity['indices'][0], entity['indices'][1]
mention_html = '<a href="https://twitter.com/%(screen_name)s" class="twython-mention">@%(screen_name)s</a>'
text = text.replace(tweet['text'][start:end], mention_html % {'screen_name': entity['screen_name']})
text = text.replace(tweet['text'][start:end],
mention_html % {'screen_name': entity['screen_name']})
# Hashtags
for entity in entities['hashtags']:
@ -495,7 +563,8 @@ class Twython(EndpointsMixin, object):
# Urls
for entity in entities['urls']:
start, end = entity['indices'][0], entity['indices'][1]
if use_display_url and entity.get('display_url') and not use_expanded_url:
if use_display_url and entity.get('display_url') \
and not use_expanded_url:
shown_url = entity['display_url']
elif use_expanded_url and entity.get('expanded_url'):
shown_url = entity['expanded_url']
@ -503,6 +572,7 @@ class Twython(EndpointsMixin, object):
shown_url = entity['url']
url_html = '<a href="%s" class="twython-url">%s</a>'
text = text.replace(tweet['text'][start:end], url_html % (entity['url'], shown_url))
text = text.replace(tweet['text'][start:end],
url_html % (entity['url'], shown_url))
return text

162
twython/endpoints.py
View file

@ -5,7 +5,8 @@ twython.endpoints
~~~~~~~~~~~~~~~~~
This module provides a mixin for a :class:`Twython <Twython>` instance.
Parameters that need to be embedded in the API url just need to be passed as a keyword argument.
Parameters that need to be embedded in the API url just need to be passed
as a keyword argument.
e.g. Twython.retweet(id=12345)
@ -20,7 +21,8 @@ class EndpointsMixin(object):
"""Returns the 20 most recent mentions (tweets containing a users's
@screen_name) for the authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/get/statuses/mentions_timeline
Docs:
https://dev.twitter.com/docs/api/1.1/get/statuses/mentions_timeline
"""
return self.get('statuses/mentions_timeline', params=params)
@ -63,7 +65,8 @@ class EndpointsMixin(object):
Docs: https://dev.twitter.com/docs/api/1.1/get/statuses/retweets/%3Aid
"""
return self.get('statuses/retweets/%s' % params.get('id'), params=params)
return self.get('statuses/retweets/%s' % params.get('id'),
params=params)
def show_status(self, **params):
"""Returns a single Tweet, specified by the id parameter
@ -73,6 +76,16 @@ class EndpointsMixin(object):
"""
return self.get('statuses/show/%s' % params.get('id'), params=params)
def lookup_status(self, **params):
"""Returns fully-hydrated tweet objects for up to 100 tweets per
request, as specified by comma-separated values passed to the id
parameter.
Docs: https://dev.twitter.com/docs/api/1.1/get/statuses/lookup
"""
return self.post('statuses/lookup', params=params)
def destroy_status(self, **params):
"""Destroys the status specified by the required ID parameter
@ -101,7 +114,8 @@ class EndpointsMixin(object):
"""Updates the authenticating user's current status and attaches media
for upload. In other words, it creates a Tweet with a picture attached.
Docs: https://dev.twitter.com/docs/api/1.1/post/statuses/update_with_media
Docs:
https://dev.twitter.com/docs/api/1.1/post/statuses/update_with_media
"""
return self.post('statuses/update_with_media', params=params)
@ -174,7 +188,8 @@ class EndpointsMixin(object):
return self.post('direct_messages/destroy', params=params)
def send_direct_message(self, **params):
"""Sends a new direct message to the specified user from the authenticating user.
"""Sends a new direct message to the specified user from the
authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/direct_messages/new
@ -186,7 +201,8 @@ class EndpointsMixin(object):
"""Returns a collection of user_ids that the currently authenticated
user does not want to receive retweets from.
Docs: https://dev.twitter.com/docs/api/1.1/get/friendships/no_retweets/ids
Docs:
https://dev.twitter.com/docs/api/1.1/get/friendships/no_retweets/ids
"""
return self.get('friendships/no_retweets/ids', params=params)
@ -317,7 +333,8 @@ class EndpointsMixin(object):
requesting user if authentication was successful; returns a 401 status
code and an error message if not.
Docs: https://dev.twitter.com/docs/api/1.1/get/account/verify_credentials
Docs:
https://dev.twitter.com/docs/api/1.1/get/account/verify_credentials
"""
return self.get('account/verify_credentials', params=params)
@ -333,13 +350,15 @@ class EndpointsMixin(object):
def update_delivery_service(self, **params):
"""Sets which device Twitter delivers updates to for the authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/update_delivery_device
Docs:
https://dev.twitter.com/docs/api/1.1/post/account/update_delivery_device
"""
return self.post('account/update_delivery_device', params=params)
def update_profile(self, **params):
"""Sets values that users are able to set under the "Account" tab of their settings page.
"""Sets values that users are able to set under the "Account" tab of their
settings page.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/update_profile
@ -349,7 +368,8 @@ class EndpointsMixin(object):
def update_profile_banner_image(self, **params): # pragma: no cover
"""Updates the authenticating user's profile background image.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/update_profile_background_image
Docs:
https://dev.twitter.com/docs/api/1.1/post/account/update_profile_background_image
"""
return self.post('account/update_profile_banner', params=params)
@ -358,7 +378,8 @@ class EndpointsMixin(object):
"""Sets one or more hex values that control the color scheme of the
authenticating user's profile page on twitter.com.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/update_profile_colors
Docs:
https://dev.twitter.com/docs/api/1.1/post/account/update_profile_colors
"""
return self.post('account/update_profile_colors', params=params)
@ -366,13 +387,15 @@ class EndpointsMixin(object):
def update_profile_image(self, **params): # pragma: no cover
"""Updates the authenticating user's profile image.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/update_profile_image
Docs:
https://dev.twitter.com/docs/api/1.1/post/account/update_profile_image
"""
return self.post('account/update_profile_image', params=params)
def list_blocks(self, **params):
"""Returns a collection of user objects that the authenticating user is blocking.
"""Returns a collection of user objects that the authenticating user
is blocking.
Docs: https://dev.twitter.com/docs/api/1.1/get/blocks/list
@ -400,7 +423,8 @@ class EndpointsMixin(object):
return self.post('blocks/create', params=params)
def destroy_block(self, **params):
"""Un-blocks the user specified in the ID parameter for the authenticating user.
"""Un-blocks the user specified in the ID parameter for the
authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/blocks/destroy
@ -409,7 +433,8 @@ class EndpointsMixin(object):
def lookup_user(self, **params):
"""Returns fully-hydrated user objects for up to 100 users per request,
as specified by comma-separated values passed to the user_id and/or screen_name parameters.
as specified by comma-separated values passed to the user_id and/or
screen_name parameters.
Docs: https://dev.twitter.com/docs/api/1.1/get/users/lookup
@ -426,7 +451,8 @@ class EndpointsMixin(object):
return self.get('users/show', params=params)
def search_users(self, **params):
"""Provides a simple, relevance-based search interface to public user accounts on Twitter.
"""Provides a simple, relevance-based search interface to public user
accounts on Twitter.
Docs: https://dev.twitter.com/docs/api/1.1/get/users/search
@ -453,7 +479,8 @@ class EndpointsMixin(object):
"""Removes the uploaded profile banner for the authenticating user.
Returns HTTP 200 upon success.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/remove_profile_banner
Docs:
https://dev.twitter.com/docs/api/1.1/post/account/remove_profile_banner
"""
return self.post('account/remove_profile_banner', params=params)
@ -461,13 +488,16 @@ class EndpointsMixin(object):
def update_profile_background_image(self, **params):
"""Uploads a profile banner on behalf of the authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/account/update_profile_banner
Docs:
https://dev.twitter.com/docs/api/1.1/post/account/update_profile_banner
"""
return self.post('account/update_profile_background_image', params=params)
return self.post('account/update_profile_background_image',
params=params)
def get_profile_banner_sizes(self, **params):
"""Returns a map of the available size variations of the specified user's profile banner.
"""Returns a map of the available size variations of the specified
user's profile banner.
Docs: https://dev.twitter.com/docs/api/1.1/get/users/profile_banner
@ -478,10 +508,12 @@ class EndpointsMixin(object):
def get_user_suggestions_by_slug(self, **params):
"""Access the users in a given category of the Twitter suggested user list.
Docs: https://dev.twitter.com/docs/api/1.1/get/users/suggestions/%3Aslug
Docs:
https://dev.twitter.com/docs/api/1.1/get/users/suggestions/%3Aslug
"""
return self.get('users/suggestions/%s' % params.get('slug'), params=params)
return self.get('users/suggestions/%s' % params.get('slug'),
params=params)
def get_user_suggestions(self, **params):
"""Access to Twitter's suggested user list.
@ -493,16 +525,20 @@ class EndpointsMixin(object):
def get_user_suggestions_statuses_by_slug(self, **params):
"""Access the users in a given category of the Twitter suggested user
list and return their most recent status if they are not a protected user.
list and return their most recent status if they are not a protected
user.
Docs: https://dev.twitter.com/docs/api/1.1/get/users/suggestions/%3Aslug/members
Docs:
https://dev.twitter.com/docs/api/1.1/get/users/suggestions/%3Aslug/members
"""
return self.get('users/suggestions/%s/members' % params.get('slug'), params=params)
return self.get('users/suggestions/%s/members' % params.get('slug'),
params=params)
# Favorites
def get_favorites(self, **params):
"""Returns the 20 most recent Tweets favorited by the authenticating or specified user.
"""Returns the 20 most recent Tweets favorited by the authenticating
or specified user.
Docs: https://dev.twitter.com/docs/api/1.1/get/favorites/list
@ -511,7 +547,8 @@ class EndpointsMixin(object):
get_favorites.iter_mode = 'id'
def destroy_favorite(self, **params):
"""Un-favorites the status specified in the ID parameter as the authenticating user.
"""Un-favorites the status specified in the ID parameter as the
authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/favorites/destroy
@ -519,7 +556,8 @@ class EndpointsMixin(object):
return self.post('favorites/destroy', params=params)
def create_favorite(self, **params):
"""Favorites the status specified in the ID parameter as the authenticating user.
"""Favorites the status specified in the ID parameter as the
authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/favorites/create
@ -528,7 +566,8 @@ class EndpointsMixin(object):
# Lists
def show_lists(self, **params):
"""Returns all lists the authenticating or specified user subscribes to, including their own.
"""Returns all lists the authenticating or specified user subscribes to,
including their own.
Docs: https://dev.twitter.com/docs/api/1.1/get/lists/list
@ -575,7 +614,8 @@ class EndpointsMixin(object):
def subscribe_to_list(self, **params):
"""Subscribes the authenticated user to the specified list.
Docs: https://dev.twitter.com/docs/api/1.1/post/lists/subscribers/create
Docs:
https://dev.twitter.com/docs/api/1.1/post/lists/subscribers/create
"""
return self.post('lists/subscribers/create', params=params)
@ -591,7 +631,8 @@ class EndpointsMixin(object):
def unsubscribe_from_list(self, **params):
"""Unsubscribes the authenticated user from the specified list.
Docs: https://dev.twitter.com/docs/api/1.1/post/lists/subscribers/destroy
Docs:
https://dev.twitter.com/docs/api/1.1/post/lists/subscribers/destroy
"""
return self.post('lists/subscribers/destroy', params=params)
@ -600,7 +641,8 @@ class EndpointsMixin(object):
"""Adds multiple members to a list, by specifying a comma-separated
list of member ids or screen names.
Docs: https://dev.twitter.com/docs/api/1.1/post/lists/members/create_all
Docs:
https://dev.twitter.com/docs/api/1.1/post/lists/members/create_all
"""
return self.post('lists/members/create_all', params=params)
@ -677,7 +719,8 @@ class EndpointsMixin(object):
"""Removes multiple members from a list, by specifying a
comma-separated list of member ids or screen names.
Docs: https://dev.twitter.com/docs/api/1.1/post/lists/members/destroy_all
Docs:
https://dev.twitter.com/docs/api/1.1/post/lists/members/destroy_all
"""
return self.post('lists/members/destroy_all', params=params)
@ -704,10 +747,12 @@ class EndpointsMixin(object):
def show_saved_search(self, **params):
"""Retrieve the information for the saved search represented by the given id.
Docs: https://dev.twitter.com/docs/api/1.1/get/saved_searches/show/%3Aid
Docs:
https://dev.twitter.com/docs/api/1.1/get/saved_searches/show/%3Aid
"""
return self.get('saved_searches/show/%s' % params.get('id'), params=params)
return self.get('saved_searches/show/%s' % params.get('id'),
params=params)
def create_saved_search(self, **params):
"""Create a new saved search for the authenticated user.
@ -720,10 +765,12 @@ class EndpointsMixin(object):
def destroy_saved_search(self, **params):
"""Destroys a saved search for the authenticating user.
Docs: https://dev.twitter.com/docs/api/1.1/post/saved_searches/destroy/%3Aid
Docs:
https://dev.twitter.com/docs/api/1.1/post/saved_searches/destroy/%3Aid
"""
return self.post('saved_searches/destroy/%s' % params.get('id'), params=params)
return self.post('saved_searches/destroy/%s' % params.get('id'),
params=params)
# Places & Geo
def get_geo_info(self, **params):
@ -851,7 +898,8 @@ class EndpointsMixin(object):
"""Returns the current rate limits for methods belonging to the
specified resource families.
Docs: https://dev.twitter.com/docs/api/1.1/get/application/rate_limit_status
Docs:
https://dev.twitter.com/docs/api/1.1/get/application/rate_limit_status
"""
return self.get('application/rate_limit_status', params=params)
@ -861,16 +909,32 @@ class EndpointsMixin(object):
TWITTER_HTTP_STATUS_CODE = {
200: ('OK', 'Success!'),
304: ('Not Modified', 'There was no new data to return.'),
400: ('Bad Request', 'The request was invalid. An accompanying error message will explain why. This is the status code will be returned during rate limiting.'),
401: ('Unauthorized', 'Authentication credentials were missing or incorrect.'),
403: ('Forbidden', 'The request is understood, but it has been refused. An accompanying error message will explain why. This code is used when requests are being denied due to update limits.'),
404: ('Not Found', 'The URI requested is invalid or the resource requested, such as a user, does not exists.'),
406: ('Not Acceptable', 'Returned by the Search API when an invalid format is specified in the request.'),
410: ('Gone', 'This resource is gone. Used to indicate that an API endpoint has been turned off.'),
422: ('Unprocessable Entity', 'Returned when an image uploaded to POST account/update_profile_banner is unable to be processed.'),
429: ('Too Many Requests', 'Returned in API v1.1 when a request cannot be served due to the application\'s rate limit having been exhausted for the resource.'),
500: ('Internal Server Error', 'Something is broken. Please post to the group so the Twitter team can investigate.'),
400: ('Bad Request', 'The request was invalid. An accompanying \
error message will explain why. This is the status code \
will be returned during rate limiting.'),
401: ('Unauthorized', 'Authentication credentials were missing \
or incorrect.'),
403: ('Forbidden', 'The request is understood, but it has been \
refused. An accompanying error message will explain why. \
This code is used when requests are being denied due to \
update limits.'),
404: ('Not Found', 'The URI requested is invalid or the resource \
requested, such as a user, does not exists.'),
406: ('Not Acceptable', 'Returned by the Search API when an \
invalid format is specified in the request.'),
410: ('Gone', 'This resource is gone. Used to indicate that an \
API endpoint has been turned off.'),
422: ('Unprocessable Entity', 'Returned when an image uploaded to \
POST account/update_profile_banner is unable to be processed.'),
429: ('Too Many Requests', 'Returned in API v1.1 when a request cannot \
be served due to the application\'s rate limit having been \
exhausted for the resource.'),
500: ('Internal Server Error', 'Something is broken. Please post to the \
group so the Twitter team can investigate.'),
502: ('Bad Gateway', 'Twitter is down or being upgraded.'),
503: ('Service Unavailable', 'The Twitter servers are up, but overloaded with requests. Try again later.'),
504: ('Gateway Timeout', 'The Twitter servers are up, but the request couldn\'t be serviced due to some failure within our stack. Try again later.'),
503: ('Service Unavailable', 'The Twitter servers are up, but overloaded \
with requests. Try again later.'),
504: ('Gateway Timeout', 'The Twitter servers are up, but the request \
couldn\'t be serviced due to some failure within our stack. Try \
again later.'),
}

2
twython/helpers.py
View file

@ -28,7 +28,7 @@ def _transparent_params(_params):
try:
params[k] = ','.join(v)
except TypeError:
params[k] = ','.join(map(str,v))
params[k] = ','.join(map(str, v))
else:
continue # pragma: no cover
return params, files

33
twython/streaming/api.py
View file

@ -38,8 +38,11 @@ class TwythonStreamer(object):
retired
:param retry_in: (optional) Amount of time (in secs) the previous
API call should be tried again
:param client_args: (optional) Accepts some requests Session parameters and some requests Request parameters.
See http://docs.python-requests.org/en/latest/api/#sessionapi and requests section below it for details.
:param client_args: (optional) Accepts some requests Session
parameters and some requests Request parameters.
See
http://docs.python-requests.org/en/latest/api/#sessionapi
and requests section below it for details.
[ex. headers, proxies, verify(SSL verification)]
:param handlers: (optional) Array of message types for which
corresponding handlers will be called
@ -53,11 +56,12 @@ class TwythonStreamer(object):
self.client_args = client_args or {}
default_headers = {'User-Agent': 'Twython Streaming v' + __version__}
if not 'headers' in self.client_args:
if 'headers' not in self.client_args:
# If they didn't set any headers, set our defaults for them
self.client_args['headers'] = default_headers
elif 'User-Agent' not in self.client_args['headers']:
# If they set headers, but didn't include User-Agent.. set it for them
# If they set headers, but didn't include User-Agent..
# set it for them
self.client_args['headers'].update(default_headers)
self.client_args['timeout'] = timeout
@ -87,7 +91,8 @@ class TwythonStreamer(object):
self.connected = False
self.handlers = handlers if handlers else ['delete', 'limit', 'disconnect']
self.handlers = handlers if handlers else \
['delete', 'limit', 'disconnect']
self.chunk_size = chunk_size
@ -103,7 +108,8 @@ class TwythonStreamer(object):
def _send(retry_counter):
requests_args = {}
for k, v in self.client_args.items():
# Maybe this should be set as a class variable and only done once?
# Maybe this should be set as a class
# variable and only done once?
if k in ('timeout', 'allow_redirects', 'verify'):
requests_args[k] = v
@ -121,7 +127,8 @@ class TwythonStreamer(object):
if response.status_code != 200:
self.on_error(response.status_code, response.content)
if self.retry_count and (self.retry_count - retry_counter) > 0:
if self.retry_count and \
(self.retry_count - retry_counter) > 0:
time.sleep(self.retry_in)
retry_counter += 1
_send(retry_counter)
@ -140,13 +147,19 @@ class TwythonStreamer(object):
line = line.decode('utf-8')
data = json.loads(line)
except ValueError: # pragma: no cover
self.on_error(response.status_code, 'Unable to decode response, not valid JSON.')
self.on_error(response.status_code,
'Unable to decode response, \
not valid JSON.')
else:
if self.on_success(data): # pragma: no cover
for message_type in self.handlers:
if message_type in data:
handler = getattr(self, 'on_' + message_type, None)
if handler and callable(handler) and not handler(data.get(message_type)):
handler = getattr(self,
'on_' + message_type,
None)
if handler \
and callable(handler) \
and not handler(data.get(message_type)):
break
response.close()