Skip to content

iterators

This module is used internally to provide utilities for iterating over paginated endpoints.

Pagination

Utility for paginated endpoints.

Attributes:

Name Type Description
_fetch_page

A callable that takes limit and offset and returns a dict with "items" and "total".

limit

The number of items to fetch per page.

Source code in luduvo/utilities/iterators.py
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class Pagination:
    """Utility for paginated endpoints.

    Attributes:
        _fetch_page: A callable that takes limit and offset and returns a dict with "items" and "total".
        limit: The number of items to fetch per page.
    """

    def __init__(
        self,
        fetch_page: Callable[[int, int], Awaitable[dict]],
        limit: int = 50,
    ):
        self._fetch_page = fetch_page
        self.limit = limit

    async def all(self) -> List[T]:
        offset = 0
        results: List[T] = []

        while True:
            data = await self._fetch_page(self.limit, offset)

            items = data["items"]
            total = data["total"]

            results.extend(items)

            offset += self.limit

            if offset >= total or not items:
                break

        return results