Сборка HTTP- и браузерных роутеров#

Корень называет доступные роутеры и выбирает для каждого клиент. MailSdk собирает HTTP-группы аккаунта и писем. BrowserMailSdk наследует их без повторного объявления и добавляет MailboxApi, который работает на странице Chromium.

Решение сборки

HTTP-запуск

Браузерный запуск

Корень

MailSdk

BrowserMailSdk

HTTP-клиент

HTTPX

curl_cffi с профилем Chrome 124

HTTP-роутеры

AccountApi, MessagesApi

те же классы роутеров

Браузерный роутер

нет

MailboxApi через bind(...)

Сессия HTTP

статический Bearer для учебного запуска

токен из mail_session после одного входа в Chromium

→ Request · GET /api/v1/user/short

Authorization: Bearer session-ada

← Response · 200 OK

{
  "status": "ok",
  "body": {"email": "ada@mail.example", "display_name": "Ada"}
}

api_group(...) публикует роутер на корне. Объявление не создаёт второй клиент и не копирует операции: группа получает клиент, Identity и сериализацию во время сборки корня.

# examples/mail/sdk.py
class AccountApi(AsyncApi):
    security = MAIL_BEARER

    current = op(GetCurrentUser)


class MessagesApi(AsyncApi):
    page = op(ListByOffset)
    send = op(SendEncryptedMessage)


class MailSdk(AsyncRoot):
    account = api_group(AccountApi)
    messages = api_group(MessagesApi)


class BrowserMailSdk(MailSdk):
    mailbox = api_group(MailboxApi)

HTTP-запуск передаёт одному MailSdk общий Identity. account.current() использует правило роутера, пагинация и зашифрованная отправка сохраняют правила своих операций.

# examples/mail/http/assembly.py
async def use_http_sdk() -> tuple[str, list[int], int]:
    site = MailSite()
    identity = Identity(
        auth=(MAIL_BEARER.static("session-1"),),
        key_provider=signing_key,
    )
    async with async_mail_client(site) as client:
        sdk = MailSdk(client, identity=identity)
        user = await sdk.account.current()
        page = await sdk.messages.page(offset=0, limit=2)
        sent = await sdk.messages.send(
            body=NewMessage(
                recipient="grace@mail.example",
                subject="Собранный SDK",
                body="Обе HTTP-группы работают.",
            )
        )
    return user.email, [message.id for message in page.items], sent.message_id or 0
Файл корня целиком
"""Final HTTP and browser roots assembled from the tutorial operations."""

from __future__ import annotations

from eazy_sdk import AsyncApi, AsyncRoot, api_group, op

from examples.mail.browser.session import MailboxApi
from examples.mail.http.encryption import SendEncryptedMessage
from examples.mail.http.messages import ListByOffset
from examples.mail.http.send import MAIL_BEARER
from examples.mail.http.session import GetCurrentUser


# region docs: mail-sdk-routers
# examples/mail/sdk.py
class AccountApi(AsyncApi):
    security = MAIL_BEARER

    current = op(GetCurrentUser)


class MessagesApi(AsyncApi):
    page = op(ListByOffset)
    send = op(SendEncryptedMessage)


class MailSdk(AsyncRoot):
    account = api_group(AccountApi)
    messages = api_group(MessagesApi)


class BrowserMailSdk(MailSdk):
    mailbox = api_group(MailboxApi)
# endregion docs: mail-sdk-routers


__all__ = ["AccountApi", "BrowserMailSdk", "MailSdk", "MessagesApi"]
Файл HTTP-запуска целиком
"""Run the assembled HTTP mail SDK against the teaching site."""

from __future__ import annotations

import asyncio

from eazy_sdk import Identity

from examples.mail.http._client import async_mail_client
from examples.mail.http.send import MAIL_BEARER, NewMessage, signing_key
from examples.mail.sdk import MailSdk
from examples.mail.site import MailSite


# region docs: http-sdk-flow
# examples/mail/http/assembly.py
async def use_http_sdk() -> tuple[str, list[int], int]:
    site = MailSite()
    identity = Identity(
        auth=(MAIL_BEARER.static("session-1"),),
        key_provider=signing_key,
    )
    async with async_mail_client(site) as client:
        sdk = MailSdk(client, identity=identity)
        user = await sdk.account.current()
        page = await sdk.messages.page(offset=0, limit=2)
        sent = await sdk.messages.send(
            body=NewMessage(
                recipient="grace@mail.example",
                subject="Собранный SDK",
                body="Обе HTTP-группы работают.",
            )
        )
    return user.email, [message.id for message in page.items], sent.message_id or 0
# endregion docs: http-sdk-flow


async def run() -> None:
    email, messages, sent = await use_http_sdk()
    print("transport: httpx")
    print("routers: account, messages")
    print(f"user: {email}")
    print(f"message ids: {messages}")
    print(f"encrypted send: {sent}")


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


if __name__ == "__main__":
    main()

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

transport: httpx
routers: account, messages
user: ada@mail.example
message ids: [42, 43]
encrypted send: 100

BrowserLogin входит один раз и отдаёт состояние контекста. Учебный сайт принимает значение куки mail_session как Bearer-токен, поэтому пример берёт его из состояния и передаёт в MAIL_BEARER.static(...) для тех же HTTP-операций. Реальный AsyncClient.curl_cffi(...) отправляет запросы, а bind(...) оставляет MailboxApi на клиенте Chromium.

Браузерный роутер ждёт data-page="mailbox", а карта писем работает с коллекцией внутри него. Обе метки приходят из фактического HTML учебного сайта:

<!-- examples/mail/site/app.py -->
<main data-page="mailbox">
  <ul data-collection="messages">

# examples/mail/browser/assembly.py
async def use_browser_sdk() -> tuple[int, str, list[int]]:
    login_service = MailLoginService()
    login = BrowserLogin(
        service=login_service,
        cookies=("mail_session",),
        expired=(SessionExpiredError,),
    )
    session = login.session(
        MailCredentials("ada@mail.example", "correct", "123456"),
        identity="ada",
    )

    async with playwright_mail() as runtime:
        browser = runtime.session_client(session)
        try:
            await MailboxApi(browser).inbox(options=OPTIONS)
            state = await session.state(browser)
            token = next(cookie.value for cookie in state.cookies if cookie.name == "mail_session")
            auth = MAIL_BEARER.static(token)
            async with mail_origin(runtime.site) as origin, AsyncClient.curl_cffi(
                base_url=origin,
                impersonate="chrome124",
            ) as http:
                sdk = BrowserMailSdk(
                    http,
                    bindings=(bind(MailboxApi, client=browser),),
                    identity=Identity(auth=(auth,), key_provider=signing_key),
                )
                user = await sdk.account.current()
                page = await sdk.messages.page(offset=0, limit=2)
                await sdk.mailbox.inbox(options=OPTIONS)
        finally:
            await browser.aclose()

    return login_service.calls, user.email, [message.id for message in page.items]

mail_origin(...) лишь поднимает локальную границу учебного MailSite. В приложении base_url указывает на сервис, а выбор impersonate="chrome124" и устройство корня остаются такими же.

Файл браузерного запуска целиком
"""Hand one Chromium login to the browser and curl_cffi routers of one root."""

from __future__ import annotations

import asyncio

from eazy_sdk import AsyncClient, Identity, bind
from eazy_sdk_browser import BrowserLogin

from examples.mail.browser.session import (
    OPTIONS,
    MailCredentials,
    MailLoginService,
    MailboxApi,
    SessionExpiredError,
)
from examples.mail.http.send import MAIL_BEARER, signing_key
from examples.mail.sdk import BrowserMailSdk
from examples.mail.site._curl import mail_origin
from examples.mail.site._playwright import playwright_mail


# region docs: browser-sdk-flow
# examples/mail/browser/assembly.py
async def use_browser_sdk() -> tuple[int, str, list[int]]:
    login_service = MailLoginService()
    login = BrowserLogin(
        service=login_service,
        cookies=("mail_session",),
        expired=(SessionExpiredError,),
    )
    session = login.session(
        MailCredentials("ada@mail.example", "correct", "123456"),
        identity="ada",
    )

    async with playwright_mail() as runtime:
        browser = runtime.session_client(session)
        try:
            await MailboxApi(browser).inbox(options=OPTIONS)
            state = await session.state(browser)
            token = next(cookie.value for cookie in state.cookies if cookie.name == "mail_session")
            auth = MAIL_BEARER.static(token)
            async with mail_origin(runtime.site) as origin, AsyncClient.curl_cffi(
                base_url=origin,
                impersonate="chrome124",
            ) as http:
                sdk = BrowserMailSdk(
                    http,
                    bindings=(bind(MailboxApi, client=browser),),
                    identity=Identity(auth=(auth,), key_provider=signing_key),
                )
                user = await sdk.account.current()
                page = await sdk.messages.page(offset=0, limit=2)
                await sdk.mailbox.inbox(options=OPTIONS)
        finally:
            await browser.aclose()

    return login_service.calls, user.email, [message.id for message in page.items]
# endregion docs: browser-sdk-flow


async def run() -> None:
    logins, email, messages = await use_browser_sdk()
    print("transport: curl_cffi + Chromium")
    print(f"browser logins: {logins}")
    print(f"user via browser session: {email}")
    print(f"message ids: {messages}")
    print("browser router: mailbox")


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


if __name__ == "__main__":
    main()

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

transport: curl_cffi + Chromium
browser logins: 1
user via browser session: ada@mail.example
message ids: [42, 43]
browser router: mailbox

Примечание

Корень не владеет клиентом, переданным через bind(...). Поэтому браузерный клиент закрывается в finally, а HTTP-клиент, созданный контекстным менеджером, закрывается своим менеджером.

Дальше#