gitea.utils.pagination

pagination

Helpers for walking paginated Gitea listings.

A listing is walked by asking for page 1, then page 2, and so on until a page says the listing has ended. What a page says is the whole difficulty: the endpoints wrapped here return a bare JSON array, so "this was the last one" has to be inferred, and an instance that infers wrongly is asked for page 3, page 4 and page 5 forever.

Four things end a walk. _end_of_listing applies all four and is the only place any of them is written down; a walker asks it about each page and does what the answer says, so the synchronous and the asynchronous walk cannot drift apart and a fifth rule is added in one place rather than two.

• The page repeats the one before it. Asked first, and the one ending whose page is not part of the listing: an instance that ignores the page parameter answers every request with the same items, so no page is ever empty, short, or different from its predecessor and nothing else below ever fires. Two consecutive identical pages cannot happen in a listing that is really being paged - the same items would have to be served twice - so the repeat is read as the listing having ended at the page before it, and its items are not handed back a second time.

The other three end the walk with the page that ended it included, because its items do belong to the listing:

• The page is empty, or shorter than the first page. The oldest signal and the one that ends almost every real listing. The length of the first page is the yardstick rather than the requested limit, because an instance may cap the page size below what was asked for.
• The response says so. page_count and has_more in a page's metadata are honoured when a caller reports them - Gitea sends the equivalent as headers on every page - and ignored when it does not, so this costs nothing to the callers that report neither. A page count has to be positive to be read as one: zero describes no listing at all, and taking it at its word would end a walk on a first page that came back full.
• The page limit. A backstop for an instance that does none of the above: one cycling through pages, say, so that no two consecutive ones match. MAX_PAGES is far above any listing these endpoints return, so reaching it means something is wrong rather than that a listing is large, and it is logged rather than quietly truncating the result.

Only the empty-or-short rule is a judgement about the data; the rest are about the instance being wrong, which is why they live here and not in each caller - every paginated command is walked through these, so none of them can be the one that still hangs.

Functions:

gitea.utils.pagination.iter_pages

iter_pages(
    fetch_page: Callable[
        [int], tuple[list[dict[str, Any]], dict[str, Any]]
    ],
) -> Iterator[tuple[list[dict[str, Any]], dict[str, Any]]]

Yield the pages of a paginated listing, one request at a time.

Pages are requested lazily, so a caller that stops early stops the requests with it.

Parameters:

Name	Type	Description	Default
fetch_page	Callable[[int], tuple[list[dict[str, Any]], dict[str, Any]]]	Callable returning the items and metadata of the given page number.	required

Yields:

Type	Description
tuple[list[dict[str, Any]], dict[str, Any]]	A tuple containing the items and the metadata of each page, in order.

Source code in src/gitea/utils/pagination.py

def iter_pages(
    fetch_page: Callable[[int], tuple[list[dict[str, Any]], dict[str, Any]]],
) -> Iterator[tuple[list[dict[str, Any]], dict[str, Any]]]:
    """Yield the pages of a paginated listing, one request at a time.

    Pages are requested lazily, so a caller that stops early stops the
    requests with it.

    Args:
        fetch_page: Callable returning the items and metadata of the given page number.

    Yields:
        A tuple containing the items and the metadata of each page, in order.

    """
    page = 1
    page_size = 0
    previous: list[dict[str, Any]] | None = None
    while True:
        batch, metadata = fetch_page(page)

        verdict = _end_of_listing(batch=batch, metadata=metadata, previous=previous, page=page, page_size=page_size)
        if verdict.include:
            yield batch, metadata
        if verdict.reason is not None:
            return

        previous = batch
        page_size = page_size or len(batch)
        page += 1

gitea.utils.pagination.iter_async_pages async

iter_async_pages(
    fetch_page: Callable[
        [int],
        Awaitable[
            tuple[list[dict[str, Any]], dict[str, Any]]
        ],
    ],
) -> AsyncIterator[
    tuple[list[dict[str, Any]], dict[str, Any]]
]

Yield the pages of a paginated listing, one request at a time.

Parameters:

Name	Type	Description	Default
fetch_page	Callable[[int], Awaitable[tuple[list[dict[str, Any]], dict[str, Any]]]]	Callable returning the items and metadata of the given page number.	required

Yields:

Type	Description
AsyncIterator[tuple[list[dict[str, Any]], dict[str, Any]]]	A tuple containing the items and the metadata of each page, in order.

Source code in src/gitea/utils/pagination.py

async def iter_async_pages(
    fetch_page: Callable[[int], Awaitable[tuple[list[dict[str, Any]], dict[str, Any]]]],
) -> AsyncIterator[tuple[list[dict[str, Any]], dict[str, Any]]]:
    """Yield the pages of a paginated listing, one request at a time.

    Args:
        fetch_page: Callable returning the items and metadata of the given page number.

    Yields:
        A tuple containing the items and the metadata of each page, in order.

    """
    page = 1
    page_size = 0
    previous: list[dict[str, Any]] | None = None
    while True:
        batch, metadata = await fetch_page(page)

        verdict = _end_of_listing(batch=batch, metadata=metadata, previous=previous, page=page, page_size=page_size)
        if verdict.include:
            yield batch, metadata
        if verdict.reason is not None:
            return

        previous = batch
        page_size = page_size or len(batch)
        page += 1

gitea.utils.pagination.collect_all_pages

collect_all_pages(
    fetch_page: Callable[
        [int], tuple[list[dict[str, Any]], dict[str, Any]]
    ],
) -> tuple[list[dict[str, Any]], dict[str, Any]]

Fetch every page of a paginated listing.

Parameters:

Name	Type	Description	Default
fetch_page	Callable[[int], tuple[list[dict[str, Any]], dict[str, Any]]]	Callable returning the items and metadata of the given page number.	required

Returns:

Type	Description
tuple[list[dict[str, Any]], dict[str, Any]]	A tuple containing every item across all pages and the metadata of the last response.

Source code in src/gitea/utils/pagination.py

def collect_all_pages(
    fetch_page: Callable[[int], tuple[list[dict[str, Any]], dict[str, Any]]],
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Fetch every page of a paginated listing.

    Args:
        fetch_page: Callable returning the items and metadata of the given page number.

    Returns:
        A tuple containing every item across all pages and the metadata of the last response.

    """
    items: list[dict[str, Any]] = []
    metadata: dict[str, Any] = {}
    for batch, page_metadata in iter_pages(fetch_page):
        items.extend(batch)
        metadata = page_metadata
    return items, metadata