Обновление сессии без ручного повтора

Сессия живёт дольше одного запроса. Страница входящих уже содержит access-токен, refresh-токен и срок действия. HTTP-клиент читает их как MailSession; браузер хранит ту же сессию в cookie контекста. Прикладной код вызывает только защищённую операцию, а вход, обновление и повтор остаются внутри жизненного цикла.

Ситуация	Что видит сайт	Что делает SDK	Результат
Первый HTTP-запрос	сессии ещё нет	читает токены из HTML	session-1
Срок почти вышел	до expires_at меньше запаса	обновляет до запроса	session-refresh-1
Access-токен отвергнут	защищённая ручка отвечает 401	обновляет и повторяет один раз	session-refresh-2
Две вкладки контекста	общая cookie mail_session	проводит один вход под локом	обе вкладки открыты

→ Request · POST /api/v1/session/refresh

{"refresh_token":"refresh-1"}

← Response · 200 OK

{"status":"ok","body":{"access_token":"session-refresh-1","refresh_token":"refresh-2","expires_at":"2030-01-01T01:05:00+00:00"}}

HTTP

Три поля одной Pydantic-модели выполняют две задачи. CSS(...) извлекает значения из HTML, а Bearer, RefreshToken и ExpiresAt описывают их роль в сессии. Запас в десять секунд не даёт отправить токен, который истечёт во время запроса.

# examples/mail/http/session.py
class MailSession(BaseModel):
    access_token: Annotated[
        SecretStr,
        Bearer(),
        CSS('meta[name="mail-token"]::attr(content)'),
    ]
    refresh_token: Annotated[
        SecretStr,
        RefreshToken(),
        CSS('meta[name="mail-refresh"]::attr(content)'),
    ]
    expires_at: Annotated[
        datetime,
        ExpiresAt(leeway=timedelta(seconds=10)),
        CSS('meta[name="mail-expires"]::attr(content)'),
    ]

При первом защищённом вызове MailSessionService.acquire читает страницу входящих. Для обновления сервис передаёт refresh-токен обычной незащищённой операции; access-токен туда не нужен и не создаёт цикл авторизации.

# examples/mail/http/session.py
@dataclass(slots=True)
class MailSessionService:
    acquired: MailSession | None = None
    refreshed: list[MailSession] = field(default_factory=list)

    async def acquire(
        self,
        credentials: str,
        context: AuthContext[MailSdk],
    ) -> MailSession:
        _ = credentials
        self.acquired = await context.sdk.session.read_from_inbox()
        return self.acquired

    async def refresh(
        self,
        session: MailSession,
        context: AuthContext[MailSdk],
    ) -> MailSession:
        renewed = await context.sdk.session.refresh(
            refresh_token=session.refresh_token.get_secret_value()
        )
        self.refreshed.append(renewed)
        return renewed

Один и тот же вызов sdk.user.current() показывает оба пути. Сначала часы пересекают локальный срок, затем сайт забывает ещё живой access-токен и отвечает 401.

# examples/mail/http/session.py
async def session_lifecycle() -> tuple[str, str, str, str, int]:
    CLOCK.now = datetime(2030, 1, 1, tzinfo=UTC)
    site = MailSite()
    service = MailSessionService()
    auth = session_auth(
        MailSession,
        credentials="inbox HTML",
        service=service,
        name="mail-session",
        clock=CLOCK,
    )
    sdk_type = protected_mail_sdk(auth.scheme)

    async with async_mail_client(site) as client:
        sdk = sdk_type(client, identity=Identity(auth=(auth,)))
        first = await sdk.user.current()

        CLOCK.now += timedelta(minutes=5)
        await sdk.user.current()
        by_expiry = service.refreshed[-1]

        site.state.sessions.pop(by_expiry.access_token.get_secret_value())
        await sdk.user.current()
        by_rejection = service.refreshed[-1]

    if service.acquired is None:
        raise RuntimeError("the protected call did not acquire a session")
    return (
        first.email,
        service.acquired.access_token.get_secret_value(),
        by_expiry.access_token.get_secret_value(),
        by_rejection.access_token.get_secret_value(),
        site.state.refreshes,
    )

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

"""Acquire a mail session from HTML, then refresh it by time and by rejection."""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Annotated

from eazy_sdk_html import CSS
from pydantic import BaseModel, SecretStr

from eazy_sdk import AsyncApi, AsyncRoot, Http, HttpOperation, Identity, JsonField, api_group, op
from eazy_sdk.auth import AuthContext, AuthScheme, Bearer, ExpiresAt, RefreshToken, session_auth
from eazy_sdk.response import ApiError, Const, Error, Html, Json, Payload

from examples.mail.http._client import async_mail_client
from examples.mail.site import MailSite


@dataclass(slots=True)
class DemoClock:
    now: datetime = field(default_factory=lambda: datetime(2030, 1, 1, tzinfo=UTC))

    def __call__(self) -> datetime:
        return self.now


CLOCK = DemoClock()


# region docs: http-session-model
# examples/mail/http/session.py
class MailSession(BaseModel):
    access_token: Annotated[
        SecretStr,
        Bearer(),
        CSS('meta[name="mail-token"]::attr(content)'),
    ]
    refresh_token: Annotated[
        SecretStr,
        RefreshToken(),
        CSS('meta[name="mail-refresh"]::attr(content)'),
    ]
    expires_at: Annotated[
        datetime,
        ExpiresAt(leeway=timedelta(seconds=10)),
        CSS('meta[name="mail-expires"]::attr(content)'),
    ]
# endregion docs: http-session-model


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


class User(BaseModel):
    email: str
    display_name: str


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


class SessionProblem(BaseModel):
    code: str
    message: str


class ProblemEnvelope(BaseModel):
    status: str
    body: Payload[SessionProblem]


class SessionRejected(ApiError[ProblemEnvelope]):
    pass


@dataclass(frozen=True, slots=True, kw_only=True)
class ReadSessionFromInbox(HttpOperation[MailSession]):
    __http__ = Http.get(
        "/inbox/",
        operation_id="readSessionFromInbox",
        success={200: Html(MailSession)},
        security=None,
    )


@dataclass(frozen=True, slots=True, kw_only=True)
class RefreshSession(HttpOperation[MailSession]):
    __http__ = Http.post(
        "/api/v1/session/refresh",
        operation_id="refreshSession",
        success={200: Json(SessionEnvelope)},
        security=None,
    )

    refresh_token: JsonField[str]


@dataclass(frozen=True, slots=True, kw_only=True)
class GetCurrentUser(HttpOperation[User]):
    __http__ = Http.get(
        "/api/v1/user/short",
        operation_id="getCurrentUser",
        success={200: Json(UserEnvelope)},
        errors=(Error(401, Json(ProblemEnvelope), exception=SessionRejected),),
    )


class SessionApi(AsyncApi):
    read_from_inbox = op(ReadSessionFromInbox)
    refresh = op(RefreshSession)


class UserApi(AsyncApi):
    current = op(GetCurrentUser)


class MailSdk(AsyncRoot):
    session = api_group(SessionApi)
    user = api_group(UserApi)


def protected_mail_sdk(scheme: AuthScheme[MailSession]) -> type[MailSdk]:
    class ProtectedUserApi(UserApi):
        security = scheme

    class ProtectedMailSdk(MailSdk):
        user = api_group(ProtectedUserApi)

    return ProtectedMailSdk


# region docs: http-session-service
# examples/mail/http/session.py
@dataclass(slots=True)
class MailSessionService:
    acquired: MailSession | None = None
    refreshed: list[MailSession] = field(default_factory=list)

    async def acquire(
        self,
        credentials: str,
        context: AuthContext[MailSdk],
    ) -> MailSession:
        _ = credentials
        self.acquired = await context.sdk.session.read_from_inbox()
        return self.acquired

    async def refresh(
        self,
        session: MailSession,
        context: AuthContext[MailSdk],
    ) -> MailSession:
        renewed = await context.sdk.session.refresh(
            refresh_token=session.refresh_token.get_secret_value()
        )
        self.refreshed.append(renewed)
        return renewed
# endregion docs: http-session-service


# region docs: http-session-flow
# examples/mail/http/session.py
async def session_lifecycle() -> tuple[str, str, str, str, int]:
    CLOCK.now = datetime(2030, 1, 1, tzinfo=UTC)
    site = MailSite()
    service = MailSessionService()
    auth = session_auth(
        MailSession,
        credentials="inbox HTML",
        service=service,
        name="mail-session",
        clock=CLOCK,
    )
    sdk_type = protected_mail_sdk(auth.scheme)

    async with async_mail_client(site) as client:
        sdk = sdk_type(client, identity=Identity(auth=(auth,)))
        first = await sdk.user.current()

        CLOCK.now += timedelta(minutes=5)
        await sdk.user.current()
        by_expiry = service.refreshed[-1]

        site.state.sessions.pop(by_expiry.access_token.get_secret_value())
        await sdk.user.current()
        by_rejection = service.refreshed[-1]

    if service.acquired is None:
        raise RuntimeError("the protected call did not acquire a session")
    return (
        first.email,
        service.acquired.access_token.get_secret_value(),
        by_expiry.access_token.get_secret_value(),
        by_rejection.access_token.get_secret_value(),
        site.state.refreshes,
    )
# endregion docs: http-session-flow


async def run() -> None:
    email, html_token, by_expiry, by_rejection, refreshes = await session_lifecycle()
    print(f"HTML session: {email} via {html_token}")
    print(f"expiry refresh: {by_expiry}")
    print(f"401 refresh: {by_rejection}")
    print(f"refresh requests: {refreshes}")


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


if __name__ == "__main__":
    main()

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

HTML session: ada@mail.example via session-1
expiry refresh: session-refresh-1
401 refresh: session-refresh-2
refresh requests: 2

Браузер

После успешного входа MailSite отдаёт страницу с таким фрагментом:

<meta name="mail-token" content="session-ada">

MailLoginService проходит уже объявленные три шага и экспортирует состояние контекста. BrowserLogin проверяет, что в нём появилась cookie mail_session; пустой результат входом не считается.

# examples/mail/browser/session.py
@dataclass(frozen=True, slots=True)
class MailCredentials:
    email: str
    password: str = field(repr=False)
    otp: str = field(repr=False)


@dataclass(slots=True)
class MailLoginService:
    calls: int = 0

    async def acquire(
        self,
        credentials: MailCredentials,
        context: BrowserLoginContext,
    ) -> BrowserState:
        self.calls += 1
        portal = LoginPortal(context.client)
        password = await portal.identify(email=credentials.email, options=OPTIONS)
        otp = await password.submit(password=credentials.password)
        await otp.submit(code=credentials.otp)
        return await context.export_state()

BrowserSession создаётся на контекст, а не на вкладку. Два клиента с разными страницами получают один объект сессии: параллельные первые переходы сходятся в один вход, затем обе вкладки используют cookie контекста.

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

    async with playwright_mail() as runtime:
        first = runtime.session_client(session)
        second = await runtime.new_tab(session)
        try:
            await asyncio.gather(
                MailboxApi(first).inbox(options=OPTIONS),
                MailboxApi(second).inbox(options=OPTIONS),
            )
            state = await session.state(first)
        finally:
            await first.aclose()
            await second.aclose()

    cookie = next(item.value for item in state.cookies if item.name == "mail_session")
    return service.calls, 2, cookie

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

"""Share one browser login between two tabs of one Chromium context."""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field

from eazy_sdk import op
from eazy_sdk_browser import (
    AsyncBrowserApi,
    Browser,
    BrowserCallOptions,
    BrowserLogin,
    BrowserLoginContext,
    BrowserOperation,
    BrowserState,
    PageError,
    css,
)

from examples.mail.browser.login_probe import LoginPortal
from examples.mail.site._playwright import playwright_mail

OPTIONS = BrowserCallOptions(timeout=5.0)


class SessionExpiredError(PageError):
    """The mail site no longer accepts the context session."""


# region docs: browser-session-login
# examples/mail/browser/session.py
@dataclass(frozen=True, slots=True)
class MailCredentials:
    email: str
    password: str = field(repr=False)
    otp: str = field(repr=False)


@dataclass(slots=True)
class MailLoginService:
    calls: int = 0

    async def acquire(
        self,
        credentials: MailCredentials,
        context: BrowserLoginContext,
    ) -> BrowserState:
        self.calls += 1
        portal = LoginPortal(context.client)
        password = await portal.identify(email=credentials.email, options=OPTIONS)
        otp = await password.submit(password=credentials.password)
        await otp.submit(code=credentials.otp)
        return await context.export_state()
# endregion docs: browser-session-login


@dataclass(frozen=True, slots=True, kw_only=True)
class OpenInbox(BrowserOperation[None, None]):
    __browser__ = Browser.goto("/inbox/", at=css('[data-page="mailbox"]'))


class MailboxApi(AsyncBrowserApi):
    inbox = op(OpenInbox)


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

    async with playwright_mail() as runtime:
        first = runtime.session_client(session)
        second = await runtime.new_tab(session)
        try:
            await asyncio.gather(
                MailboxApi(first).inbox(options=OPTIONS),
                MailboxApi(second).inbox(options=OPTIONS),
            )
            state = await session.state(first)
        finally:
            await first.aclose()
            await second.aclose()

    cookie = next(item.value for item in state.cookies if item.name == "mail_session")
    return service.calls, 2, cookie
# endregion docs: browser-session-flow


async def run() -> None:
    logins, tabs, cookie = await one_login_for_two_tabs()
    print(f"browser logins: {logins}")
    print(f"tabs in context: {tabs}")
    print(f"session cookie: {cookie}")


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


if __name__ == "__main__":
    main()

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

browser logins: 1
tabs in context: 2
session cookie: session-ada

Примечание

У BrowserSession один объект на один контекст. Для второго контекста создайте второй объект через login.session(...); общее хранилище может согласовать их ревизии, но не объединяет cookie jars.

Дальше

• Прочитайте список писем по частям.

• Посмотрите, как сессия появилась после входа.