Войти и получить сессию#

Управляемая сессия выполняет login перед первым защищённым вызовом. Приложение передаёт LoginCredentials, а LoginService вызывает обычную объявленную операцию и возвращает UserSession.

→ Запрос · POST /auth/login

{"email": "ada@mail.example", "password": "correct-horse"}

← Ответ · 200 OK

{
  "access_token": "access-1",
  "refresh_token": "refresh-1",
  "expires_at": "2099-01-01T00:00:00Z"
}

Модель отмечает роль каждого поля сессии. session_scheme(...) создаёт схему, которую можно поставить в security роутера.

# examples/docs/auth_lifecycle.py
class LoginCredentials(BaseModel):
    email: str
    password: SecretStr


class UserSession(BaseModel):
    access_token: Annotated[SecretStr, Bearer()]
    refresh_token: Annotated[SecretStr, RefreshToken()]
    expires_at: Annotated[datetime, ExpiresAt()]


class Account(BaseModel):
    id: str
    email: str


USER_SESSION = session_scheme(UserSession, name="mail-session")

Login и refresh остаются обычными публичными операциями, а защищённый роутер использует USER_SESSION по умолчанию.

# examples/docs/auth_lifecycle.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Login(HttpOperation[UserSession]):
    __http__ = Http.post("/auth/login", security=None)

    email: JsonField[str]
    password: JsonField[str]


@dataclass(frozen=True, slots=True, kw_only=True)
class Refresh(HttpOperation[UserSession]):
    __http__ = Http.post("/auth/refresh", security=None)

    refresh_token: JsonField[str]


@dataclass(frozen=True, slots=True, kw_only=True)
class GetAccount(HttpOperation[Account]):
    __http__ = Http.get("/account", errors={401: SessionRejected})


class AuthApi(AsyncApi):
    login = op(Login)
    refresh = op(Refresh)


class AccountApi(AsyncApi):
    security = USER_SESSION
    get_account = op(GetAccount)

acquire(...) получает типизированный context.sdk. Корень объявляет обе группы один раз.

# examples/docs/auth_lifecycle.py
class LoginService:
    async def acquire(
        self,
        credentials: LoginCredentials,
        context: AuthContext[MailSdk],
    ) -> UserSession:
        return await context.sdk.auth.login(
            email=credentials.email,
            password=credentials.password.get_secret_value(),
        )

    async def refresh(
        self,
        session: UserSession,
        context: AuthContext[MailSdk],
    ) -> UserSession:
        return await context.sdk.auth.refresh(
            refresh_token=session.refresh_token.get_secret_value(),
        )


class MailSdk(AsyncRoot):
    auth = api_group(AuthApi)
    account = api_group(AccountApi)

Первый защищённый вызов запускает acquire(...), сохраняет результат и повторно планирует операцию уже с Authorization.

# examples/docs/auth_login.py
async def main() -> None:
    site = AuthSite()
    credentials = LoginCredentials(
        email="ada@mail.example",
        password=SecretStr("correct-horse"),
    )
    async with mail_sdk(site, credentials=credentials) as sdk:
        account = await sdk.account.get_account()

    print(f"account: {account.id} {account.email}")
    print(f"first request: {site.calls[0]}")

Браузерный вход начинается с реальной HTML-формы, а пошаговые операции BrowserLogin разобраны в главе учебного примера.

<!-- examples/mail/site/app.py -->
<main data-page="identify">
  <form method="post" action="/login/identify">
    <label>Почта <input name="email" type="email"></label>
    <button type="submit">Продолжить</button>
  </form>
</main>

Ответ login

Правило

Что получает вызывающий

200 с моделью сессии

acquire(...) завершён

Результат защищённого метода

Отображённый ApiError

Сессия не сохранена

Типизированное исключение

Неизвестный или повреждённый ответ

Разбор модели не выполнен

Ошибка протокола
Запускаемый файл целиком
"""Acquire a session automatically before a protected request."""

import asyncio

from pydantic import SecretStr

from examples.docs.auth_lifecycle import AuthSite, LoginCredentials, mail_sdk


# region docs: auth-login-run
# examples/docs/auth_login.py
async def main() -> None:
    site = AuthSite()
    credentials = LoginCredentials(
        email="ada@mail.example",
        password=SecretStr("correct-horse"),
    )
    async with mail_sdk(site, credentials=credentials) as sdk:
        account = await sdk.account.get_account()

    print(f"account: {account.id} {account.email}")
    print(f"first request: {site.calls[0]}")
# endregion docs: auth-login-run


if __name__ == "__main__":
    asyncio.run(main())

account: user-42 ada@mail.example
first request: POST /auth/login

session_auth(...) остаётся короткой формой для SDK без общей схемы роутера. В корне с несколькими группами используйте session_scheme(...).configure(...), как в этом примере.

Дальше#