Обойти страницы результата#

Стратегия пагинации принадлежит классу операции. Обычный вызов возвращает одну страницу, а pages() и items() строят следующие запросы через неизменяемое значение операции.

# examples/docs/pagination.py
@dataclass(frozen=True, slots=True, kw_only=True)
class ListDocuments(HttpOperation[DocumentPage]):
    __http__ = Http.get("/documents")
    __pages__ = Pages.numbered(
        DocumentPage,
        page="page",
        size="per_page",
        items=lambda result: result.items,
        total_pages=lambda result: result.pages_count,
    )

    page: Query[int] = 1
    per_page: Query[int] = 2


class DocumentsApi(SyncApi):
    list = op(ListDocuments)


def main() -> None:
    with guide_client(GuideSite()) as client:
        documents = DocumentsApi(client)
        for number, page in enumerate(documents.list.pages(), start=1):
            print(f"page {number}: {', '.join(page.items)}")

page= и size= называют поля Python-класса, а не параметры на проводе. items= извлекает элементы, total_pages= сообщает последнюю страницу. Несовпадающий тип результата или опечатка в имени поля останавливает объявление роутера при импорте.

Файл целиком
"""Traverse numbered pages declared on an operation."""

from dataclasses import dataclass

from pydantic import BaseModel

from eazy_sdk import Http, HttpOperation, Query, SyncApi, op
from eazy_sdk.pagination import Pages
from examples.docs.guide_site import GuideSite, guide_client


class DocumentPage(BaseModel):
    items: list[str]
    pages_count: int


# region docs: pagination
# examples/docs/pagination.py
@dataclass(frozen=True, slots=True, kw_only=True)
class ListDocuments(HttpOperation[DocumentPage]):
    __http__ = Http.get("/documents")
    __pages__ = Pages.numbered(
        DocumentPage,
        page="page",
        size="per_page",
        items=lambda result: result.items,
        total_pages=lambda result: result.pages_count,
    )

    page: Query[int] = 1
    per_page: Query[int] = 2


class DocumentsApi(SyncApi):
    list = op(ListDocuments)


def main() -> None:
    with guide_client(GuideSite()) as client:
        documents = DocumentsApi(client)
        for number, page in enumerate(documents.list.pages(), start=1):
            print(f"page {number}: {', '.join(page.items)}")
# endregion docs: pagination


if __name__ == "__main__":
    main()

page 1: Иск, Отзыв
page 2: Решение

Выбрать стратегию#

Стратегия

Поле следующего запроса

Признак окончания

Pages.numbered

Номер страницы

Общее число страниц, короткая или пустая страница

Pages.offset

Смещение

Общее число элементов, короткая или пустая страница

Pages.cursor

Cursor из ответа

Отсутствующий следующий cursor

Pages.next_url

Готовая ссылка

Отсутствующая ссылка

items() разворачивает страницы в один поток. Аргумент key= пропускает повторяющиеся элементы и завершает обход, если новая страница не добавила ни одного значения. max_pages= ограничивает число запросов, а options= применяется к каждой странице.

max_items= ограничивает число элементов: items(max_items=5) отдаёт первые пять и не запрашивает страницы после той, на которой нашёлся пятый. Повторы, пропущенные по key=, в счёт не идут.

first() возвращает первый элемент или None, если первая страница пуста, и всегда делает ровно один запрос: latest = await api.documents.first(case_id=cid).

Тип элемента берётся из items= стратегии: если items=lambda r: r.items возвращает list[Document], то items() отдаёт Document, first() возвращает Document | None, а key= принимает Document. Отдельная аннотация в вызывающем коде не нужна.

Каждая страница проходит обычные retry, middleware, авторизацию, защиту и подпись. Синхронный роутер возвращает iterator, асинхронный async iterator.

Дальше#