Чтение всех писем по частям#

Почта отдаёт по два письма. HTTP-клиент может следовать смещению, курсору или готовой ссылке; браузер видит первую пару в HTML и просит страницу подгрузить вторую. Во всех четырёх случаях результат один: письма 42, 43, 44 и 45 в исходном порядке.

Вариант	Адрес следующей порции	Условие остановки
HTTP, `Pages.offset`	`offset + len(items)`	достигнут `total`
HTTP, `Pages.cursor`	`next_cursor` из ответа	курсор равен `null`
HTTP, `Pages.next_url`	готовый `next_url`	ссылка равна `null`
Браузер, `each(...)`	кнопка `data-action="more"`	четвёртый элемент появился в коллекции

→ Request · GET /api/v1/threads/status/smart?offset=0&limit=2

Authorization: Bearer session-1

← Response · 200 OK

{
  "status": "ok",
  "body": {
    "items": [
      {"id": 42, "sender": "ada@mail.example", "recipient": "reader@mail.example", "subject": "Планы на пятницу"},
      {"id": 43, "sender": "grace@mail.example", "recipient": "reader@mail.example", "subject": "Сверка протокола"}
    ],
    "total": 4,
    "next_offset": 2,
    "next_cursor": "cursor-2",
    "next_url": "/api/v1/threads/status/smart?offset=2&limit=2"
  }
}

HTTP

Три операции обращаются к одной ручке и разбирают одну модель ответа. Различается только __pages__: стратегия называет поля запроса и функцию, которая читает указатель следующей порции.

# examples/mail/http/messages.py
@dataclass(frozen=True, slots=True, kw_only=True)
class ListByOffset(HttpOperation[MessagePage]):
    __http__ = Http.get(
        "/api/v1/threads/status/smart",
        success={200: Json(MessagePageEnvelope)},
        security=MAIL_BEARER,
    )
    __pages__ = Pages.offset(
        MessagePage,
        offset="offset",
        limit="limit",
        items=lambda page: page.items,
        total=lambda page: page.total,
    )

    offset: Query[int] = 0
    limit: Query[int] = 2


@dataclass(frozen=True, slots=True, kw_only=True)
class ListByCursor(HttpOperation[MessagePage]):
    __http__ = Http.get(
        "/api/v1/threads/status/smart",
        success={200: Json(MessagePageEnvelope)},
        security=MAIL_BEARER,
    )
    __pages__ = Pages.cursor(
        MessagePage,
        cursor="cursor",
        items=lambda page: page.items,
        next_cursor=lambda page: page.next_cursor,
    )

    cursor: Query[Omittable[str]] = UNSET
    limit: Query[int] = 2


@dataclass(frozen=True, slots=True, kw_only=True)
class ListByNextUrl(HttpOperation[MessagePage]):
    __http__ = Http.get(
        "/api/v1/threads/status/smart",
        success={200: Json(MessagePageEnvelope)},
        security=MAIL_BEARER,
    )
    __pages__ = Pages.next_url(
        MessagePage,
        items=lambda page: page.items,
        next_url=lambda page: page.next_url,
    )

    offset: Query[int] = 0
    limit: Query[int] = 2


class MessagesApi(SyncApi):
    base_url = "https://mail.example"

    by_offset = op(ListByOffset)
    by_cursor = op(ListByCursor)
    by_next_url = op(ListByNextUrl)

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

# examples/mail/http/messages.py
def read_with_three_strategies() -> tuple[list[int], list[int], list[int]]:
    identity = Identity(auth=(MAIL_BEARER.static("session-1"),))
    with mail_client() as client:
        api = MessagesApi(client, identity=identity)
        offset = [message.id for message in api.by_offset.items()]
        cursor = [message.id for message in api.by_cursor.items()]
        next_url = [message.id for message in api.by_next_url.items()]
    return offset, cursor, next_url

Файл примера целиком

"""Read the same mail feed with offset, cursor and next-URL pagination."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated

from pydantic import BaseModel

from eazy_sdk import UNSET, Http, HttpOperation, Identity, Omittable, Query, SyncApi, op
from eazy_sdk.pagination import Pages
from eazy_sdk.response import Const, Json, Payload

from examples.mail.http._client import mail_client
from examples.mail.http.send import MAIL_BEARER


class Message(BaseModel):
    id: int
    sender: str
    recipient: str
    subject: str


class MessagePage(BaseModel):
    items: list[Message]
    total: int
    next_offset: int | None
    next_cursor: str | None
    next_url: str | None


class MessagePageEnvelope(BaseModel):
    status: Annotated[str, Const("ok")]
    body: Payload[MessagePage]


# region docs: http-message-pages
# examples/mail/http/messages.py
@dataclass(frozen=True, slots=True, kw_only=True)
class ListByOffset(HttpOperation[MessagePage]):
    __http__ = Http.get(
        "/api/v1/threads/status/smart",
        success={200: Json(MessagePageEnvelope)},
        security=MAIL_BEARER,
    )
    __pages__ = Pages.offset(
        MessagePage,
        offset="offset",
        limit="limit",
        items=lambda page: page.items,
        total=lambda page: page.total,
    )

    offset: Query[int] = 0
    limit: Query[int] = 2


@dataclass(frozen=True, slots=True, kw_only=True)
class ListByCursor(HttpOperation[MessagePage]):
    __http__ = Http.get(
        "/api/v1/threads/status/smart",
        success={200: Json(MessagePageEnvelope)},
        security=MAIL_BEARER,
    )
    __pages__ = Pages.cursor(
        MessagePage,
        cursor="cursor",
        items=lambda page: page.items,
        next_cursor=lambda page: page.next_cursor,
    )

    cursor: Query[Omittable[str]] = UNSET
    limit: Query[int] = 2


@dataclass(frozen=True, slots=True, kw_only=True)
class ListByNextUrl(HttpOperation[MessagePage]):
    __http__ = Http.get(
        "/api/v1/threads/status/smart",
        success={200: Json(MessagePageEnvelope)},
        security=MAIL_BEARER,
    )
    __pages__ = Pages.next_url(
        MessagePage,
        items=lambda page: page.items,
        next_url=lambda page: page.next_url,
    )

    offset: Query[int] = 0
    limit: Query[int] = 2


class MessagesApi(SyncApi):
    base_url = "https://mail.example"

    by_offset = op(ListByOffset)
    by_cursor = op(ListByCursor)
    by_next_url = op(ListByNextUrl)
# endregion docs: http-message-pages


# region docs: http-message-flow
# examples/mail/http/messages.py
def read_with_three_strategies() -> tuple[list[int], list[int], list[int]]:
    identity = Identity(auth=(MAIL_BEARER.static("session-1"),))
    with mail_client() as client:
        api = MessagesApi(client, identity=identity)
        offset = [message.id for message in api.by_offset.items()]
        cursor = [message.id for message in api.by_cursor.items()]
        next_url = [message.id for message in api.by_next_url.items()]
    return offset, cursor, next_url
# endregion docs: http-message-flow


def main() -> None:
    offset, cursor, next_url = read_with_three_strategies()
    print(f"offset: {offset}")
    print(f"cursor: {cursor}")
    print(f"next URL: {next_url}")


if __name__ == "__main__":
    main()

Запуск: uv run python -m examples.mail.http.messages.

offset: [42, 43, 44, 45]
cursor: [42, 43, 44, 45]
next URL: [42, 43, 44, 45]

Браузер

Первая страница входящих содержит два элемента и кнопку:

<ul data-collection="messages"><li data-message-id="42"><b>ada@mail.example</b> Планы на пятницу</li><li data-message-id="43"><b>grace@mail.example</b> Сверка протокола</li></ul><button data-action="more">Ещё</button>

each(...) связывает поле карты со всеми видимыми li. Сначала операция читает снимок коллекции, затем нажимает кнопку и ждёт элемент с индексом 3. После этого новый снимок уже содержит обе порции.

# examples/mail/browser/messages.py
class InboxPage:
    messages: Annotated[Elements, each(css(MESSAGE))]
    more: Annotated[Element, css('[data-action="more"]')]


@dataclass(frozen=True, slots=True)
class LoadedMessages:
    first_batch: tuple[int, ...]
    all_messages: tuple[int, ...]


async def _ids(messages: Elements) -> tuple[int, ...]:
    values = [await item.attribute("data-message-id") async for item in messages]
    return tuple(int(value) for value in values if value is not None)


async def _load_more(content: InboxPage) -> LoadedMessages:
    first_batch = await _ids(content.messages)
    await content.more.click()
    await content.messages.nth(3).text()
    return LoadedMessages(first_batch, await _ids(content.messages))


async def _mailbox_missing(content: InboxPage) -> NoReturn:
    _ = content
    raise TimeoutError("the inbox did not appear")


@dataclass(frozen=True, slots=True, kw_only=True)
class ReadInbox(BrowserOperation[InboxPage, LoadedMessages]):
    __browser__ = Browser.goto(
        "/inbox/",
        InboxPage,
        at=css(MESSAGE),
        outcomes=outcomes(
            when(visible(css(MESSAGE)), then=_load_more),
            otherwise=_mailbox_missing,
        ),
    )


class InboxApi(AsyncBrowserApi):
    read = op(ReadInbox)

Вызов открывает /inbox/ в настоящем Chromium. JavaScript страницы отправляет второй запрос к тому же MailSite и добавляет полученные узлы в список.

# examples/mail/browser/messages.py
async def read_all_messages() -> LoadedMessages:
    async with playwright_mail() as runtime:
        return await InboxApi(runtime.client).read(options=OPTIONS)

Файл примера целиком

"""Read the initial inbox collection and load the next batch in Chromium."""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from typing import Annotated, NoReturn

from eazy_sdk import op
from eazy_sdk_browser import (
    AsyncBrowserApi,
    Browser,
    BrowserCallOptions,
    BrowserOperation,
    Element,
    Elements,
    css,
    each,
    outcomes,
    visible,
    when,
)

from examples.mail.site._playwright import playwright_mail

OPTIONS = BrowserCallOptions(timeout=5.0, element_timeout=5.0)
MESSAGE = '[data-collection="messages"] li'


# region docs: browser-message-collection
# examples/mail/browser/messages.py
class InboxPage:
    messages: Annotated[Elements, each(css(MESSAGE))]
    more: Annotated[Element, css('[data-action="more"]')]


@dataclass(frozen=True, slots=True)
class LoadedMessages:
    first_batch: tuple[int, ...]
    all_messages: tuple[int, ...]


async def _ids(messages: Elements) -> tuple[int, ...]:
    values = [await item.attribute("data-message-id") async for item in messages]
    return tuple(int(value) for value in values if value is not None)


async def _load_more(content: InboxPage) -> LoadedMessages:
    first_batch = await _ids(content.messages)
    await content.more.click()
    await content.messages.nth(3).text()
    return LoadedMessages(first_batch, await _ids(content.messages))


async def _mailbox_missing(content: InboxPage) -> NoReturn:
    _ = content
    raise TimeoutError("the inbox did not appear")


@dataclass(frozen=True, slots=True, kw_only=True)
class ReadInbox(BrowserOperation[InboxPage, LoadedMessages]):
    __browser__ = Browser.goto(
        "/inbox/",
        InboxPage,
        at=css(MESSAGE),
        outcomes=outcomes(
            when(visible(css(MESSAGE)), then=_load_more),
            otherwise=_mailbox_missing,
        ),
    )


class InboxApi(AsyncBrowserApi):
    read = op(ReadInbox)
# endregion docs: browser-message-collection


# region docs: browser-message-flow
# examples/mail/browser/messages.py
async def read_all_messages() -> LoadedMessages:
    async with playwright_mail() as runtime:
        return await InboxApi(runtime.client).read(options=OPTIONS)
# endregion docs: browser-message-flow


async def run() -> None:
    messages = await read_all_messages()
    print(f"first batch: {list(messages.first_batch)}")
    print(f"after more: {list(messages.all_messages)}")


def main() -> None:
    asyncio.run(run())


if __name__ == "__main__":
    main()

Запуск: uv run python -m examples.mail.browser.messages.

first batch: [42, 43]
after more: [42, 43, 44, 45]

Примечание

Elements.count() и Elements.texts() читают снимок без ожидания. Когда нужен будущий узел, обращайтесь к nth(i): действие над ним ждёт появления элемента в пределах таймаута.

Чтение всех писем по частям#

Дальше#