Source code for redditwarp.pagination.paginator


from __future__ import annotations
from typing import TypeVar, Sequence, Generic, Iterator, Optional

T = TypeVar('T')

# Abstract classes

[docs]class Paginator(Generic[T]):
    def __init__(self, *, limit: Optional[int] = None) -> None:
        self.limit: Optional[int] = limit
        ("")

    def __iter__(self) -> Iterator[Sequence[T]]:
        while page := self.fetch():
            yield page

[docs]    def fetch(self) -> Sequence[T]:
        raise NotImplementedError

[docs]class OffsetPaginator(Paginator[T]):
    def __init__(self, *, limit: Optional[int] = None, offset: Optional[int] = None) -> None:
        super().__init__(limit=limit)
        self.offset: Optional[int] = offset
        ("")

[docs]class CursorPaginator(Paginator[T]):
[docs]    def get_cursor(self) -> str:
        raise NotImplementedError

[docs]    def set_cursor(self, value: str) -> None:
        raise NotImplementedError


# Interfaces

[docs]class Bidirectional:
    direction: bool

[docs]class Resettable:
[docs]    def reset(self) -> None:
        raise NotImplementedError

[docs]class HasMore:
[docs]    def has_more(self) -> bool:
        raise NotImplementedError

[docs]    def set_has_more(self, value: bool) -> None:
        raise NotImplementedError


# Mixins

[docs]class HasMorePaginator(HasMore, Paginator[T]):
    def __iter__(self) -> Iterator[Sequence[T]]:
        if page := self.fetch():
            yield page
            while self.has_more() and (page := self.fetch()):
                yield page