Source code for cybsi.cloud.pagination

"""
Pagination API.

See Also:
    See :ref:`pagination-example`
    for complete examples of pagination usage.
"""
from typing import (
    AsyncIterator,
    Callable,
    Coroutine,
    Generic,
    Iterator,
    List,
    Optional,
    TypeVar,
    cast,
)
from urllib.parse import parse_qs, urlparse

import httpx


[docs] class Cursor: """Page cursor value. If :data:`None` cursor value is **passed** to SDK (as a parameter to filter-like functions), SDK retrieves the first page. Typically, SDK does not **return** cursors as function return values. SDK returns :class:`Page`, and cursor is a property of the page. :data:`None` cursor value of a page means last page. """ pass
# This is a hack to prevent Sphinx autodoc-typehint type inlining. # If we simply alias Cursor = str, it inlines str everywhere, # and functions lose descriptive parameter and return value types. # Additionally, this hack prevents SDK users from creating Cursor instances. # Users have to call filter()-like methods. Cursor.__supertype__ = str # type: ignore T = TypeVar("T") class _BasePage(Generic[T]): def __init__(self, resp: httpx.Response, view: Callable[..., T]): self._resp = resp self._view = view @property def next_link(self) -> str: """Next page link.""" return cast(str, self._resp.links.get("next", {}).get("url")) @property def cursor(self) -> Optional[Cursor]: """Page cursor. The current position in the collection. :data:`None` means the page is last one. """ next_url = self._resp.links.get("next", {}).get("url") if next_url is None: return None parsed = urlparse(next_url) query = parse_qs(parsed.query) cursor = query.get("cursor") return cast(Optional[Cursor], cursor[0]) if cursor is not None else None def data(self) -> List[T]: """Get page data as a list of items.""" return list(iter(self)) def __iter__(self) -> Iterator[T]: yield from (self._view(x) for x in self._resp.json())
[docs] class Page(_BasePage[T]): """Page returned by Cybsi Cloud API. Should not be constructed manually, use filter-like methods provided by SDK. Args: api_call: Callable object for getting next page resp: Response which represents a start page view: View class for page elements """ def __init__( self, api_call: Callable[..., httpx.Response], resp: httpx.Response, view: Callable[..., T], ): super().__init__(resp, view) self._api_call = api_call
[docs] def next_page(self) -> "Optional[Page[T]]": """Get next page. If there is no link to the next page it return None. """ if self.next_link is None: return None return Page(self._api_call, self._api_call(self.next_link), self._view)
[docs] class AsyncPage(_BasePage[T]): """Page returned by Cybsi API. Should not be constructed manually, use filter-like methods provided by SDK. Args: api_call: Callable object for getting next page resp: Response which represents a start page view: View class for page elements """ def __init__( self, api_call: Callable[..., Coroutine], resp: httpx.Response, view: Callable[..., T], ): super().__init__(resp, view) self._api_call = api_call
[docs] async def next_page(self) -> "Optional[AsyncPage[T]]": """Get next page. If there is no link to the next page it return None. """ if self.next_link is None: return None resp = await self._api_call(self.next_link) return AsyncPage(self._api_call, resp, self._view)
[docs] def chain_pages(start_page: Page[T]) -> Iterator[T]: """Get chain of collection objects.""" page: Optional[Page[T]] = start_page while page: yield from page page = page.next_page()
[docs] async def chain_pages_async(start_page: AsyncPage[T]) -> AsyncIterator[T]: """Get chain of collection objects asynchronously.""" page: Optional[AsyncPage[T]] = start_page while page: for elem in page: yield elem page = await page.next_page()