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

Merge pull request #209 from ryanmcgrath/dev

Browse source 
2.10.1
This commit is contained in:
Mike Helmick 2013-05-29 08:42:31 -07:00
commit 6e1dc6b9bc
11 changed files with 229 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.travis.yml
View file

@ -26,7 +26,7 @@ env:
- PROTECTED_TWITTER_2=TwythonSecure2
- TEST_TWEET_ID=332992304010899457
- TEST_LIST_ID=574
script: nosetests -v test_twython:TwythonAPITestCase test_twython:TwythonAuthTestCase --logging-filter="twython" --cover-package="twython" --with-coverage
script: nosetests -v test_twython:TwythonAPITestCase test_twython:TwythonAuthTestCase test_twython:TwythonStreamTestCase --logging-filter="twython" --cover-package="twython" --with-coverage
install: pip install -r requirements.txt
notifications:
email: false

2
AUTHORS.rst
View file

@ -40,3 +40,5 @@ Patches and Suggestions
- `Greg Nofi <https://github.com/nofeet>`_, fixed using built-in Exception attributes for storing & retrieving error message
- `Jonathan Vanasco <https://github.com/jvanasco>`_, Debugging support, error_code tracking, Twitter error API tracking, other fixes
- `DevDave <https://github.com/devdave>`_, quick fix for longs with helper._transparent_params
- `Ruben Varela Rosa <https://github.com/rubenvarela>`_, Fixed search example
>>>>>>> Update stream example, update AUTHORS for future example fix

11
HISTORY.rst
View file

@ -3,6 +3,17 @@
History
-------
2.10.1 (2013-05-xx)
++++++++++++++++++
- More test coverage!
- Fix ``search_gen``
- Fixed ``get_lastfunction_header`` to actually do what its docstring says, returns ``None`` if header is not found
- Updated some internal API code, ``__init__`` didn't need to have ``self.auth`` and ``self.headers`` because they were never used anywhere else but the ``__init__``
- Added ``disconnect`` method to ``TwythonStreamer``, allowing users to disconnect as they desire
- Updated ``TwythonStreamError`` docstring, also allow importing it from ``twython``
- No longer raise ``TwythonStreamError`` when stream line can't be decoded. Instead, sends signal to ``TwythonStreamer.on_error``
- Allow for (int, long, float) params to be passed to Twython Twitter API functions in Python 2, and (int, float) in Python 3
2.10.0 (2013-05-21)
++++++++++++++++++
- Added ``get_retweeters_ids`` method

5
examples/stream.py
View file

@ -3,7 +3,10 @@ from twython import TwythonStreamer
class MyStreamer(TwythonStreamer):
def on_success(self, data):
print data
if 'text' in data:
print data['text'].encode('utf-8')
# Want to disconnect after the first result?
# self.disconnect()
def on_error(self, status_code, data):
print status_code, data

4
setup.py
View file

@ -1,10 +1,12 @@
#!/usr/bin/env python
import os
import sys
from setuptools import setup
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
__version__ = '2.10.0'
__version__ = '2.10.1'
packages = [
'twython',

145
test_twython.py
View file

@ -1,7 +1,11 @@
import unittest
import os
from twython import(
Twython, TwythonStreamer, TwythonError,
TwythonAuthError, TwythonStreamError
)
from twython import Twython, TwythonError, TwythonAuthError
import os
import time
import unittest
app_key = os.environ.get('APP_KEY')
app_secret = os.environ.get('APP_SECRET')
@ -24,6 +28,7 @@ test_list_id = os.environ.get('TEST_LIST_ID', '574') # 574 is @twitter/team
class TwythonAuthTestCase(unittest.TestCase):
def setUp(self):
self.api = Twython(app_key, app_secret)
self.bad_api = Twython('BAD_APP_KEY', 'BAD_APP_SECRET')
def test_get_authentication_tokens(self):
'''Test getting authentication tokens works'''
@ -31,11 +36,83 @@ class TwythonAuthTestCase(unittest.TestCase):
force_login=True,
screen_name=screen_name)
def test_get_authentication_tokens_bad_tokens(self):
'''Test getting authentication tokens with bad tokens
raises TwythonAuthError'''
self.assertRaises(TwythonAuthError, self.bad_api.get_authentication_tokens,
callback_url='http://google.com/')
def test_get_authorized_tokens_bad_tokens(self):
'''Test getting final tokens fails with wrong tokens'''
self.assertRaises(TwythonError, self.bad_api.get_authorized_tokens,
'BAD_OAUTH_VERIFIER')
class TwythonAPITestCase(unittest.TestCase):
def setUp(self):
self.api = Twython(app_key, app_secret,
oauth_token, oauth_token_secret)
oauth_token, oauth_token_secret,
headers={'User-Agent': '__twython__ Test'})
def test_construct_api_url(self):
'''Test constructing a Twitter API url works as we expect'''
url = 'https://api.twitter.com/1.1/search/tweets.json'
constructed_url = self.api.construct_api_url(url, {'q': '#twitter'})
self.assertEqual(constructed_url, 'https://api.twitter.com/1.1/search/tweets.json?q=%23twitter')
def test_shorten_url(self):
'''Test shortening a url works'''
self.api.shorten_url('http://google.com')
def test_shorten_url_no_shortner(self):
'''Test shortening a url with no shortener provided raises TwythonError'''
self.assertRaises(TwythonError, self.api.shorten_url,
'http://google.com', '')
def test_get(self):
'''Test Twython generic GET request works'''
self.api.get('account/verify_credentials')
def test_post(self):
'''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!'})
self.api.post('statuses/destroy/%s' % status['id_str'])
def test_get_lastfunction_header(self):
'''Test getting last specific header of the last API call works'''
self.api.get('statuses/home_timeline')
self.api.get_lastfunction_header('x-rate-limit-remaining')
def test_get_lastfunction_header_not_present(self):
'''Test getting specific header that does not exist from the last call returns None'''
self.api.get('statuses/home_timeline')
header = self.api.get_lastfunction_header('does-not-exist')
self.assertEqual(header, None)
def test_get_lastfunction_header_no_last_api_call(self):
'''Test attempting to get a header when no API call was made raises a TwythonError'''
self.assertRaises(TwythonError, self.api.get_lastfunction_header,
'no-api-call-was-made')
def test_search_gen(self):
'''Test looping through the generator results works, at least once that is'''
search = self.api.search_gen('twitter', count=1)
counter = 0
while counter < 2:
counter += 1
result = next(search)
new_id_str = int(result['id_str'])
if counter == 1:
prev_id_str = new_id_str
time.sleep(1) # Give time for another tweet to come into search
if counter == 2:
self.assertTrue(new_id_str > prev_id_str)
def test_encode(self):
'''Test encoding UTF-8 works'''
self.api.encode('Twython is awesome!')
# Timelines
def test_get_mentions_timeline(self):
@ -77,20 +154,6 @@ class TwythonAPITestCase(unittest.TestCase):
status = self.api.update_status(status='Test post just to get deleted :(')
self.api.destroy_status(id=status['id_str'])
def test_retweet(self):
'''Test retweeting a status succeeds'''
retweet = self.api.retweet(id='99530515043983360')
self.api.destroy_status(id=retweet['id_str'])
def test_retweet_twice(self):
'''Test that trying to retweet a tweet twice raises a TwythonError'''
retweet = self.api.retweet(id='99530515043983360')
self.assertRaises(TwythonError, self.api.retweet,
id='99530515043983360')
# Then clean up
self.api.destroy_status(id=retweet['id_str'])
def test_get_oembed_tweet(self):
'''Test getting info to embed tweet on Third Party site succeeds'''
self.api.get_oembed_tweet(id='99530515043983360')
@ -117,7 +180,7 @@ class TwythonAPITestCase(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!')
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'])
@ -132,7 +195,7 @@ class TwythonAPITestCase(unittest.TestCase):
def test_get_user_ids_of_blocked_retweets(self):
'''Test that collection of user_ids that the authenticated user does
not want to receive retweets from succeeds'''
self.api.get_user_ids_of_blocked_retweets(stringify_ids='true')
self.api.get_user_ids_of_blocked_retweets(stringify_ids=True)
def test_get_friends_ids(self):
'''Test returning ids of users the authenticated user and then a random
@ -412,5 +475,47 @@ class TwythonAPITestCase(unittest.TestCase):
self.api.get_closest_trends(lat='37', long='-122')
class TwythonStreamTestCase(unittest.TestCase):
def setUp(self):
class MyStreamer(TwythonStreamer):
def on_success(self, data):
self.disconnect()
def on_error(self, status_code, data):
raise TwythonStreamError(data)
def on_delete(self, data):
return
def on_limit(self, data):
return
def on_disconnect(self, data):
return
def on_timeout(self, data):
return
self.api = MyStreamer(app_key, app_secret,
oauth_token, oauth_token_secret)
def test_stream_status_filter(self):
self.api.statuses.filter(track='twitter')
def test_stream_status_sample(self):
self.api.statuses.sample()
def test_stream_status_firehose(self):
self.assertRaises(TwythonStreamError, self.api.statuses.firehose,
track='twitter')
def test_stream_site(self):
self.assertRaises(TwythonStreamError, self.api.site,
follow='twitter')
def test_stream_user(self):
self.api.user(track='twitter')
if __name__ == '__main__':
unittest.main()

7
twython/__init__.py
View file

@ -18,8 +18,11 @@ Questions, comments? ryan@venodesigns.net
"""
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
__version__ = '2.10.0'
__version__ = '2.10.1'
from .twython import Twython
from .streaming import TwythonStreamer
from .exceptions import TwythonError, TwythonRateLimitError, TwythonAuthError
from .exceptions import (
TwythonError, TwythonRateLimitError, TwythonAuthError,
TwythonStreamError
)

14
twython/exceptions.py
View file

@ -2,8 +2,7 @@ from .endpoints import twitter_http_status_codes
class TwythonError(Exception):
"""
Generic error class, catch-all for most Twython issues.
"""Generic error class, catch-all for most Twython issues.
Special cases are handled by TwythonAuthError & TwythonRateLimitError.
Note: Syntax has changed as of Twython 1.3. To catch these,
@ -11,8 +10,7 @@ class TwythonError(Exception):
from twython import (
TwythonError, TwythonRateLimitError, TwythonAuthError
)
"""
)"""
def __init__(self, msg, error_code=None, retry_after=None):
self.error_code = error_code
@ -31,8 +29,7 @@ class TwythonError(Exception):
class TwythonAuthError(TwythonError):
"""Raised when you try to access a protected resource and it fails due to
some issue with your authentication.
"""
some issue with your authentication."""
pass
@ -40,8 +37,7 @@ class TwythonRateLimitError(TwythonError):
"""Raised when you've hit a rate limit.
The amount of seconds to retry your request in will be appended
to the message.
"""
to the message."""
def __init__(self, msg, error_code, retry_after=None):
if isinstance(retry_after, int):
msg = '%s (Retry after %d seconds)' % (msg, retry_after)
@ -49,5 +45,5 @@ class TwythonRateLimitError(TwythonError):
class TwythonStreamError(TwythonError):
"""Test"""
"""Raised when an invalid response from the Stream API is received"""
pass

22
twython/streaming/api.py
View file

@ -1,5 +1,5 @@
from .. import __version__
from ..compat import json
from ..compat import json, is_py3
from ..exceptions import TwythonStreamError
from .types import TwythonStreamerTypes
@ -13,6 +13,7 @@ class TwythonStreamer(object):
def __init__(self, app_key, app_secret, oauth_token, oauth_token_secret,
timeout=300, retry_count=None, retry_in=10, headers=None):
"""Streaming class for a friendly streaming user experience
Authentication IS required to use the Twitter Streaming API
:param app_key: (required) Your applications key
:param app_secret: (required) Your applications secret key
@ -55,14 +56,18 @@ class TwythonStreamer(object):
self.user = StreamTypes.user
self.site = StreamTypes.site
self.connected = False
def _request(self, url, method='GET', params=None):
"""Internal stream request handling"""
self.connected = True
retry_counter = 0
method = method.lower()
func = getattr(self.client, method)
def _send(retry_counter):
while self.connected:
try:
if method == 'get':
response = func(url, params=params, timeout=self.timeout)
@ -81,15 +86,23 @@ class TwythonStreamer(object):
return response
while self.connected:
response = _send(retry_counter)
for line in response.iter_lines():
if not self.connected:
break
if line:
try:
if not is_py3:
self.on_success(json.loads(line))
else:
line = line.decode('utf-8')
self.on_success(json.loads(line))
except ValueError:
raise TwythonStreamError('Response was not valid JSON, \
unable to decode.')
self.on_error(response.status_code, 'Unable to decode response, not vaild JSON.')
response.close()
def on_success(self, data):
"""Called when data has been successfull received from the stream
@ -161,3 +174,6 @@ class TwythonStreamer(object):
def on_timeout(self):
return
def disconnect(self):
self.connected = False

2
twython/streaming/types.py
View file

@ -61,7 +61,7 @@ class TwythonStreamerTypesStatuses(object):
self.streamer._request(url, params=params)
def firehose(self, **params):
"""Stream statuses/filter
"""Stream statuses/firehose
Accepted params found at:
https://dev.twitter.com/docs/api/1.1/get/statuses/firehose

38
twython/twython.py
View file

@ -59,27 +59,27 @@ class Twython(object):
stacklevel=2
)
self.headers = {'User-Agent': 'Twython v' + __version__}
req_headers = {'User-Agent': 'Twython v' + __version__}
if headers:
self.headers.update(headers)
req_headers.update(headers)
# Generate OAuth authentication object for the request
# If no keys/tokens are passed to __init__, self.auth=None allows for
# If no keys/tokens are passed to __init__, auth=None allows for
# unauthenticated requests, although I think all v1.1 requests need auth
self.auth = None
auth = None
if self.app_key is not None and self.app_secret is not None and \
self.oauth_token is None and self.oauth_token_secret is None:
self.auth = OAuth1(self.app_key, self.app_secret)
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.auth = OAuth1(self.app_key, self.app_secret,
auth = OAuth1(self.app_key, self.app_secret,
self.oauth_token, self.oauth_token_secret)
self.client = requests.Session()
self.client.headers = self.headers
self.client.headers = req_headers
self.client.proxies = proxies
self.client.auth = self.auth
self.client.auth = auth
self.client.verify = ssl_verify
# register available funcs to allow listing name when debugging.
@ -208,7 +208,7 @@ class Twython(object):
def request(self, endpoint, method='GET', params=None, version='1.1'):
# In case they want to pass a full Twitter URL
# i.e. https://search.twitter.com/
# i.e. https://api.twitter.com/1.1/search/tweets.json
if endpoint.startswith('http://') or endpoint.startswith('https://'):
url = endpoint
else:
@ -241,9 +241,11 @@ class Twython(object):
"""
if self._last_call is None:
raise TwythonError('This function must be called after an API call. It delivers header information.')
if header in self._last_call['headers']:
return self._last_call['headers'][header]
return self._last_call
else:
return None
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
@ -325,7 +327,7 @@ class Twython(object):
stacklevel=2
)
if shortener == '':
if not shortener:
raise TwythonError('Please provide a URL shortening service.')
request = requests.get(shortener, params={
@ -336,7 +338,7 @@ class Twython(object):
if request.status_code in [301, 201, 200]:
return request.text
else:
raise TwythonError('shortenURL() failed with a %s error code.' % request.status_code)
raise TwythonError('shorten_url failed with a %s error code.' % request.status_code)
@staticmethod
def constructApiURL(base_url, params):
@ -373,25 +375,25 @@ class Twython(object):
See Twython.search() for acceptable parameters
e.g search = x.searchGen('python')
e.g search = x.search_gen('python')
for result in search:
print result
"""
kwargs['q'] = search_query
content = self.search(q=search_query, **kwargs)
if not content['results']:
if not content.get('statuses'):
raise StopIteration
for tweet in content['results']:
for tweet in content['statuses']:
yield tweet
try:
kwargs['page'] = 2 if not 'page' in kwargs else (int(kwargs['page']) + 1)
if not 'since_id' in kwargs:
kwargs['since_id'] = (int(content['statuses'][0]['id_str']) + 1)
except (TypeError, ValueError):
raise TwythonError('Unable to generate next page of search results, `page` is not a number.')
for tweet in self.searchGen(search_query, **kwargs):
for tweet in self.search_gen(search_query, **kwargs):
yield tweet
@staticmethod