Source code for globus_sdk.paging.marker

from __future__ import annotations

import typing as t

from .base import PageT, Paginator


[docs] class MarkerPaginator(Paginator[PageT]): """ A paginator which uses `has_next_page` and `marker` from payloads, sets the `marker` query param to page. This is the default method for GCS pagination, so it's very simple. """ _REQUIRES_METHOD_KWARGS = ("marker",) def __init__( self, method: t.Callable[..., t.Any], *, items_key: str | None = None, marker_key: str = "marker", client_args: tuple[t.Any, ...], client_kwargs: dict[str, t.Any], ): super().__init__( method, items_key=items_key, client_args=client_args, client_kwargs=client_kwargs, ) self.marker: str | None = None self.marker_key = marker_key def _check_has_next_page(self, page: dict[str, t.Any]) -> bool: return bool(page.get("has_next_page", False))
[docs] def pages(self) -> t.Iterator[PageT]: has_next_page = True while has_next_page: if self.marker: self.client_kwargs["marker"] = self.marker current_page = self.method(*self.client_args, **self.client_kwargs) yield current_page self.marker = current_page.get(self.marker_key) has_next_page = self._check_has_next_page(current_page)
class NullableMarkerPaginator(MarkerPaginator[PageT]): """ A paginator which uses a ``marker`` from payloads and sets the ``marker`` query param to page. Unlike the base MarkerPaginator, it checks for a null marker to indicate an end to pagination. (vs an explicit has_next_page key) """ def _check_has_next_page(self, page: dict[str, t.Any]) -> bool: return page.get(self.marker_key) is not None