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

Merge branch 'master' of github.com:ryanmcgrath/twython into cursor-pages

Browse source 
Conflicts:
	twython/api.py
This commit is contained in:
drevicko 2014-03-05 13:30:28 +11:00
commit 144c60c3e5
2 changed files with 44 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

11
tests/test_core.py
View file

@ -277,6 +277,17 @@ class TwythonAPITestCase(unittest.TestCase):
"""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')
@responses.activate
def test_sends_correct_accept_encoding_header(self):
"""Test that Twython accepts compressed data."""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.get(endpoint)
self.assertEqual(b'gzip, deflate, compress', responses.calls[0].request.headers['Accept-Encoding'])
# Static methods
def test_construct_api_url(self):
"""Test constructing a Twitter API url works as we expect"""

64
twython/api.py
View file

@ -109,10 +109,14 @@ class Twython(EndpointsMixin, object):
# Never be used again.
client_args_copy = self.client_args.copy()
for k, v in client_args_copy.items():
if k in ('cert', 'headers', 'hooks', 'max_redirects', 'proxies'):
if k in ('cert', 'hooks', 'max_redirects', 'proxies'):
setattr(self.client, k, v)
self.client_args.pop(k) # Pop, pop!
# Headers are always present, so we unconditionally pop them and merge
# them into the session headers.
self.client.headers.update(self.client_args.pop('headers'))
self._last_call = None
def __repr__(self):
@ -406,41 +410,39 @@ class Twython(EndpointsMixin, object):
if not hasattr(function, 'iter_mode'):
raise TwythonError('Unable to create generator for Twython method "%s"' % function.__name__)
content = function(**params)
while True:
content = function(**params)
if not content:
raise StopIteration
if not content:
raise StopIteration
if hasattr(function, 'iter_key'):
results = content.get(function.iter_key)
else:
results = content
if hasattr(function, 'iter_key'):
results = content.get(function.iter_key)
else:
results = content
if return_pages:
yield results
else:
for result in results:
yield result
if return_pages:
yield results
else:
for result in results:
yield result
if function.iter_mode == 'cursor' and content['next_cursor_str'] == '0':
raise StopIteration
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 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 result in self.cursor(function, return_pages=return_pages, **params):
yield result
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 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.')
@staticmethod
def unicode2utf8(text):