from math import inf
+import requests
+
+from tweepy.client import Response
+
class AsyncPaginator:
""":class:`AsyncPaginator` can be used to paginate for any
.. versionadded:: 4.11
+ .. note::
+
+ When passing ``return_type=requests.Response`` to :class:`Client` for
+ pagination, payload of response will be deserialized implicitly to get
+ ``meta`` attribute every requests, which may affect performance.
+
Parameters
----------
method
class AsyncPaginationIterator:
-
def __init__(
- self, method, *args, limit=inf, pagination_token=None, reverse=False,
- **kwargs
+ self, method, *args, limit=inf, pagination_token=None, reverse=False, **kwargs
):
self.method = method
self.args = args
# https://twittercommunity.com/t/why-does-timeline-use-pagination-token-while-search-uses-next-token/150963
if self.method.__name__ in (
- "search_all_tweets", "search_recent_tweets",
- "get_all_tweets_count"
+ "search_all_tweets",
+ "search_recent_tweets",
+ "get_all_tweets_count",
):
self.kwargs["next_token"] = pagination_token
else:
response = await self.method(*self.args, **self.kwargs)
- self.previous_token = response.meta.get("previous_token")
- self.next_token = response.meta.get("next_token")
+ if isinstance(response, Response):
+ meta = response.meta
+ elif isinstance(response, dict):
+ meta = response.get("meta", {})
+ elif isinstance(response, requests.Response):
+ meta = response.json().get("meta", {})
+ else:
+ raise NotImplementedError(
+ f"Unknown {type(response)} return type for {self.method}"
+ )
+
+ self.previous_token = meta.get("previous_token")
+ self.next_token = meta.get("next_token")
self.count += 1
return response
# Copyright 2009-2022 Joshua Roesslein
# See LICENSE for details.
-from typing import Union
from math import inf
import requests
.. versionadded:: 4.0
+ .. note::
+
+ When passing ``return_type=requests.Response`` to :class:`Client` for
+ pagination, payload of response will be deserialized implicitly to get
+ ``meta`` attribute every requests, which may affect performance.
+
Parameters
----------
method
return PaginationIterator(self.method, *self.args, **self.kwargs)
def __reversed__(self):
- return PaginationIterator(self.method, *self.args, reverse=True,
- **self.kwargs)
+ return PaginationIterator(self.method, *self.args, reverse=True, **self.kwargs)
def flatten(self, limit=inf):
"""Flatten paginated data
return
count = 0
- for response in PaginationIterator(self.method, *self.args,
- **self.kwargs):
+ for response in PaginationIterator(self.method, *self.args, **self.kwargs):
if response.data is not None:
for data in response.data:
yield data
class PaginationIterator:
-
- def __init__(self, method, *args, limit=inf, pagination_token=None,
- reverse=False, **kwargs):
+ def __init__(
+ self, method, *args, limit=inf, pagination_token=None, reverse=False, **kwargs
+ ):
self.method = method
self.args = args
self.limit = limit
def __iter__(self):
return self
- def __next__(self) -> Union[Response, dict, requests.Response]:
+ def __next__(self):
if self.reverse:
pagination_token = self.previous_token
else:
# https://twittercommunity.com/t/why-does-timeline-use-pagination-token-while-search-uses-next-token/150963
if self.method.__name__ in (
- "search_all_tweets", "search_recent_tweets",
- "get_all_tweets_count"
+ "search_all_tweets",
+ "search_recent_tweets",
+ "get_all_tweets_count",
):
self.kwargs["next_token"] = pagination_token
else:
elif isinstance(response, requests.Response):
meta = response.json().get("meta", {})
else:
- raise NotImplementedError("Unknown `response` type to parse `meta` field")
+ raise NotImplementedError(
+ f"Unknown {type(response)} return type for {self.method}"
+ )
self.previous_token = meta.get("previous_token")
self.next_token = meta.get("next_token")