Source code for redditwarp.pagination.async_paginator


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

T = TypeVar('T')

# Abstract classes

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

    async def __aiter__(self) -> AsyncIterator[Sequence[T]]:
        while page := await self.fetch():
            yield page

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

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

[docs]class CursorAsyncPaginator(AsyncPaginator[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 HasMoreAsyncPaginator(HasMore, AsyncPaginator[T]):
    async def __aiter__(self) -> AsyncIterator[Sequence[T]]:
        if page := await self.fetch():
            yield page
            while self.has_more() and (page := await self.fetch()):
                yield page