Source code for groupy.pagers
from groupy import utils
[docs]class Pager:
"""Class for iterating over multiple pages of results.
This is a generic, base class. To create a specific type of pager, provide
a definition for ``set_next_page_params`` in a subclass.
:param manager: the manager from which to get results
:type manager: :class:`~groupy.api.base.Manager`
:param func endpoint: a callable from which results can be fetched
:param kwargs params: initial params to pass to the manager
"""
#: the base set of params
default_params = {}
def __init__(self, manager, endpoint, **params):
self.manager = manager
self.endpoint = endpoint
params = {k: v for k, v in params.items() if v is not None}
self.params = dict(self.default_params, **params)
self.items = self.fetch()
def __getitem__(self, index):
return self.items[index]
def __iter__(self):
return iter(self.items)
[docs] def set_next_page_params(self):
"""Set the params in preparation for fetching the next page."""
raise NotImplementedError
[docs] def fetch(self):
"""Fetch the current page of results.
:return: the current page of results
:rtype: :class:`list`
"""
return self.endpoint(**self.params)
[docs] def fetch_next(self):
"""Fetch the next page of results.
:return: the next page of results
:rtype: :class:`list`
"""
self.set_next_page_params()
return self.fetch()
[docs] def autopage(self):
"""Iterate through results from all pages.
:return: all results
:rtype: generator
"""
while self.items:
yield from self.items
self.items = self.fetch_next()
[docs]class GroupList(Pager):
"""Pager for groups."""
#: default to the first page
default_params = {'page': 1}
def set_next_page_params(self):
self.params['page'] += 1
class ChatList(GroupList):
pass
[docs]class MessageList(Pager):
"""Pager for messages."""
#: the default mode param
default_mode = 'before_id'
#: all possible mode params and the index for their next page params
modes = {
'before_id': -1,
'after_id': -1,
'since_id': 0,
}
def __init__(self, manager, endpoint, **params):
super().__init__(manager, endpoint, **params)
self.mode = self.__class__.detect_mode(**params)
[docs] @classmethod
def detect_mode(cls, **params):
"""Detect which listing mode of the given params.
:params kwargs params: the params
:return: one of the available modes
:rtype: str
:raises ValueError: if multiple modes are detected
"""
modes = []
for mode in cls.modes:
if params.get(mode) is not None:
modes.append(mode)
if len(modes) > 1:
error_message = 'ambiguous mode, must be one of {}'
modes_csv = ', '.join(list(cls.modes))
raise ValueError(error_message.format(modes_csv))
return modes[0] if modes else cls.default_mode
[docs] def set_next_page_params(self):
"""Set the params so that the next page is fetched."""
if self.items:
index = self.get_last_item_index()
self.params[self.mode] = self.get_next_page_param(self.items[index])
[docs] def get_last_item_index(self):
"""Return the index of the last item in the page."""
return self.modes[self.mode]
[docs] def get_next_page_param(self, item):
"""Return the param from the given item.
:param item: the item that has the next page param
:returns: next page param value
"""
return item.id
def fetch_next(self):
return super().fetch_next() if self.modes[self.mode] else []
[docs]class GalleryList(MessageList):
"""Pager for gallery messages."""
default_mode = 'before'
modes = {
'before': -1,
'after': -1,
'since': 0,
}
def get_next_page_param(self, item):
return utils.get_rfc3339(item.created_at)