Вход в три шага#
Вход не стоит прятать в один метод. После адреса сайт показывает форму пароля, после пароля переводит на второй фактор и только после кода выдаёт сессию. Отдельная операция на каждый шаг сохраняет этот порядок в типах.
Шаг |
Запрос или страница |
Следующее состояние |
|---|---|---|
Адрес |
|
|
Пароль |
|
|
Код |
|
|
→ Request · POST /login/identify
{"email": "ada@mail.example"}
← Response · 200 OK
{
"status": "ok",
"body": {"login_id": "login:ada@mail.example"}
}
Каждый HTTP-вызов получает конверт с status и body. Payload снимает внешний конверт,
поэтому операция возвращает сразу модель следующего шага.
# examples/mail/http/login_probe.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Identify(HttpOperation[PasswordStep]):
__http__ = Http.post("/login/identify", success={200: Json(PasswordStepResponse)})
email: JsonField[str]
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitPassword(HttpOperation[OtpStep]):
__http__ = Http.post("/login/password", success={200: Json(OtpStepResponse)})
login_id: JsonField[str]
password: JsonField[str]
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitOtp(HttpOperation[Session]):
__http__ = Http.post("/login/otp", success={200: Json(SessionResponse)})
login_id: JsonField[str]
code: JsonField[str]
Роутер публикует три операции и наследует общие варианты отказа от LoginService.
# examples/mail/http/login_probe.py
class LoginApi(LoginService, SyncApi):
identify = op(Identify)
password = op(SubmitPassword)
otp = op(SubmitOtp)
В прикладном коде результат предыдущего вызова даёт данные для следующего:
# examples/mail/http/login_success.py
def login(login_api: LoginApi) -> str:
password_step = login_api.identify(email="ada@mail.example")
otp_step = login_api.password(
login_id=password_step.login_id,
password=VALID_INPUT,
)
session = login_api.otp(login_id=otp_step.login_id, code="123456")
return session.access_token
Файлы примера целиком
"""Probe a three-step login declaration and its six documented outcomes."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Annotated
from eazy_sdk import Http, HttpOperation, JsonField, SyncApi, op
from eazy_sdk.response import ApiError, Const, Error, Json, Payload
BASE_URL = "https://mail.example"
WRONG_INPUT = "wrong"
VALID_INPUT = "correct"
@dataclass(frozen=True, slots=True)
class PasswordStep:
login_id: str
@dataclass(frozen=True, slots=True)
class OtpStep:
login_id: str
destination: str
@dataclass(frozen=True, slots=True)
class Session:
access_token: str
@dataclass(frozen=True, slots=True)
class PasswordStepResponse:
status: Annotated[str, Const("ok")]
body: Payload[PasswordStep]
@dataclass(frozen=True, slots=True)
class OtpStepResponse:
status: Annotated[str, Const("ok")]
body: Payload[OtpStep]
@dataclass(frozen=True, slots=True)
class SessionResponse:
status: Annotated[str, Const("ok")]
body: Payload[Session]
# region docs: http-login-failures
# examples/mail/http/login_probe.py
@dataclass(frozen=True, slots=True)
class WrongPasswordDetails:
code: Annotated[str, Const("wrong_password")]
message: str
@dataclass(frozen=True, slots=True)
class AccountNotFoundDetails:
code: Annotated[str, Const("account_not_found")]
message: str
@dataclass(frozen=True, slots=True)
class AccountBlockedDetails:
code: Annotated[str, Const("account_blocked")]
message: str
@dataclass(frozen=True, slots=True)
class WrongCodeDetails:
code: Annotated[str, Const("wrong_code")]
message: str
@dataclass(frozen=True, slots=True)
class WrongPasswordResponse:
status: Annotated[str, Const("wrong_password")]
body: Payload[WrongPasswordDetails]
@dataclass(frozen=True, slots=True)
class AccountNotFoundResponse:
status: Annotated[str, Const("account_not_found")]
body: Payload[AccountNotFoundDetails]
@dataclass(frozen=True, slots=True)
class AccountBlockedResponse:
status: Annotated[str, Const("account_blocked")]
body: Payload[AccountBlockedDetails]
@dataclass(frozen=True, slots=True)
class WrongCodeResponse:
status: Annotated[str, Const("wrong_code")]
body: Payload[WrongCodeDetails]
@dataclass(frozen=True, slots=True)
class CaptchaRequiredResponse:
kind: Annotated[str, Const("captcha")]
site_key: str
class WrongPassword(ApiError[WrongPasswordResponse]):
pass
class AccountNotFound(ApiError[AccountNotFoundResponse]):
pass
class AccountBlocked(ApiError[AccountBlockedResponse]):
pass
class WrongCode(ApiError[WrongCodeResponse]):
pass
class CaptchaRequired(ApiError[CaptchaRequiredResponse]):
pass
class LoginService:
"""Failures shared by all three operations of the login service."""
base_url = BASE_URL
errors = (
Error(200, Json(WrongPasswordResponse), exception=WrongPassword),
Error(200, Json(AccountNotFoundResponse), exception=AccountNotFound),
Error(200, Json(AccountBlockedResponse), exception=AccountBlocked),
Error(200, Json(WrongCodeResponse), exception=WrongCode),
Error(403, Json(CaptchaRequiredResponse), exception=CaptchaRequired),
)
# endregion docs: http-login-failures
# region docs: http-login-operations
# examples/mail/http/login_probe.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Identify(HttpOperation[PasswordStep]):
__http__ = Http.post("/login/identify", success={200: Json(PasswordStepResponse)})
email: JsonField[str]
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitPassword(HttpOperation[OtpStep]):
__http__ = Http.post("/login/password", success={200: Json(OtpStepResponse)})
login_id: JsonField[str]
password: JsonField[str]
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitOtp(HttpOperation[Session]):
__http__ = Http.post("/login/otp", success={200: Json(SessionResponse)})
login_id: JsonField[str]
code: JsonField[str]
# endregion docs: http-login-operations
# region docs: http-login-router
# examples/mail/http/login_probe.py
class LoginApi(LoginService, SyncApi):
identify = op(Identify)
password = op(SubmitPassword)
otp = op(SubmitOtp)
# endregion docs: http-login-router
"""Run the successful three-step HTTP login from the tutorial."""
from __future__ import annotations
from examples.mail.http._client import mail_client
from examples.mail.http.login_probe import VALID_INPUT, LoginApi
# region docs: http-login-flow
# examples/mail/http/login_success.py
def login(login_api: LoginApi) -> str:
password_step = login_api.identify(email="ada@mail.example")
otp_step = login_api.password(
login_id=password_step.login_id,
password=VALID_INPUT,
)
session = login_api.otp(login_id=otp_step.login_id, code="123456")
return session.access_token
# endregion docs: http-login-flow
def main() -> None:
with mail_client() as client:
token = login(LoginApi(client))
print(f"session: {token}")
if __name__ == "__main__":
main()
Запуск: uv run python -m examples.mail.http.login_success.
session: session-ada
Браузерные операции заполняют по одной форме. После клика правило when(...) ждёт признак
следующей страницы, а to= строит состояние с тем же роутером.
После отправки адреса тот же MailSite возвращает страницу пароля:
<main data-page="password">
<form method="post" action="/login/password">
<input name="login_id" type="hidden" value="login:ada@mail.example">
<label>Пароль <input name="password" type="password"></label>
<button type="submit">Войти</button>
</form>
</main>
# examples/mail/browser/login_probe.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Identify(BrowserOperation[IdentifyPage, PasswordStep]):
__browser__ = Browser.act(
IdentifyPage,
at=css('input[name="email"]'),
outcomes=outcomes(
when(visible(css('input[name="password"]')), to=PasswordStep),
otherwise=_identify_stalled,
),
)
email: str
async def act(self, content: IdentifyPage) -> None:
await content.email.fill(self.email)
await content.submit.click()
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitPassword(BrowserOperation[PasswordPage, OtpStep]):
__browser__ = Browser.act(
PasswordPage,
at=css('input[name="password"]'),
errors=PASSWORD_FAILURES,
outcomes=outcomes(
when(visible(css('input[name="code"]')), to=OtpStep),
otherwise=_password_stalled,
),
)
password: str
async def act(self, content: PasswordPage) -> None:
await content.password.fill(self.password)
await content.submit.click()
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitOtp(BrowserOperation[OtpPage, Mailbox]):
__browser__ = Browser.act(
OtpPage,
at=css('input[name="code"]'),
errors=OTP_FAILURES,
outcomes=outcomes(
when(visible(css('[data-page="mailbox"]')), to=Mailbox),
otherwise=_otp_stalled,
),
)
code: str
async def act(self, content: OtpPage) -> None:
await content.code.fill(self.code)
await content.submit.click()
# examples/mail/browser/login_probe.py
class LoginPortal(AsyncBrowserApi):
errors = ACCOUNT_FAILURES
identify = op(Identify)
password = op(SubmitPassword)
otp = op(SubmitOtp)
Цепочка ниже проходит обычный mypy-гейт. Если операция вернёт не тот шаг, присваивание не сойдётся по типу.
# examples/mail/browser/login_probe.py
async def typed_login(portal: LoginPortal) -> Mailbox:
"""Make loss of a step type fail the ordinary mypy gate."""
password: PasswordStep = await portal.identify(email="ada@mail.example")
otp: OtpStep = await password.submit(password=VALID_LOGIN_INPUT)
mailbox: Mailbox = await otp.submit(code="123456")
return mailbox
Файл примера целиком
"""Type probe for a three-step browser login expressed with ``to=`` states."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Annotated, NoReturn
from eazy_sdk_browser import (
AsyncBrowserApi,
Browser,
BrowserOperation,
Element,
Failure,
PageError,
css,
outcomes,
visible,
when,
)
from eazy_sdk import op
VALID_LOGIN_INPUT = "correct"
class WrongPasswordError(PageError):
"""Пароль не подошёл."""
class AccountNotFoundError(PageError):
"""Аккаунт не найден."""
class AccountBlockedError(PageError):
"""Аккаунт заблокирован."""
class WrongCodeError(PageError):
"""Код подтверждения не подошёл."""
class CaptchaRequiredError(PageError):
"""Страница запросила капчу."""
# region docs: browser-login-failures
# examples/mail/browser/login_probe.py
ACCOUNT_FAILURES = (
Failure(
when=visible(css('[data-code="account_not_found"]')),
exception=AccountNotFoundError,
),
Failure(
when=visible(css('[data-code="account_blocked"]')),
exception=AccountBlockedError,
),
)
PASSWORD_FAILURES = (
Failure(
when=visible(css('[data-code="wrong_password"]')),
exception=WrongPasswordError,
),
Failure(
when=visible(css('[data-page="captcha"]')),
exception=CaptchaRequiredError,
),
)
OTP_FAILURES = (
Failure(
when=visible(css('[data-code="wrong_code"]')),
exception=WrongCodeError,
),
)
# endregion docs: browser-login-failures
class IdentifyPage:
email: Annotated[Element, css('input[name="email"]')]
submit: Annotated[Element, css('button[type="submit"]')]
class PasswordPage:
password: Annotated[Element, css('input[name="password"]')]
submit: Annotated[Element, css('button[type="submit"]')]
class OtpPage:
code: Annotated[Element, css('input[name="code"]')]
submit: Annotated[Element, css('button[type="submit"]')]
@dataclass(frozen=True, slots=True)
class Mailbox:
api: LoginPortal
@dataclass(frozen=True, slots=True)
class PasswordStep:
api: LoginPortal
async def submit(self, password: str) -> OtpStep:
return await self.api.password(password=password)
@dataclass(frozen=True, slots=True)
class OtpStep:
api: LoginPortal
async def submit(self, code: str) -> Mailbox:
return await self.api.otp(code=code)
async def _identify_stalled(content: IdentifyPage) -> NoReturn:
_ = content
raise TimeoutError("password step did not appear")
async def _password_stalled(content: PasswordPage) -> NoReturn:
_ = content
raise TimeoutError("OTP step did not appear")
async def _otp_stalled(content: OtpPage) -> NoReturn:
_ = content
raise TimeoutError("mailbox did not appear")
# region docs: browser-login-operations
# examples/mail/browser/login_probe.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Identify(BrowserOperation[IdentifyPage, PasswordStep]):
__browser__ = Browser.act(
IdentifyPage,
at=css('input[name="email"]'),
outcomes=outcomes(
when(visible(css('input[name="password"]')), to=PasswordStep),
otherwise=_identify_stalled,
),
)
email: str
async def act(self, content: IdentifyPage) -> None:
await content.email.fill(self.email)
await content.submit.click()
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitPassword(BrowserOperation[PasswordPage, OtpStep]):
__browser__ = Browser.act(
PasswordPage,
at=css('input[name="password"]'),
errors=PASSWORD_FAILURES,
outcomes=outcomes(
when(visible(css('input[name="code"]')), to=OtpStep),
otherwise=_password_stalled,
),
)
password: str
async def act(self, content: PasswordPage) -> None:
await content.password.fill(self.password)
await content.submit.click()
@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitOtp(BrowserOperation[OtpPage, Mailbox]):
__browser__ = Browser.act(
OtpPage,
at=css('input[name="code"]'),
errors=OTP_FAILURES,
outcomes=outcomes(
when(visible(css('[data-page="mailbox"]')), to=Mailbox),
otherwise=_otp_stalled,
),
)
code: str
async def act(self, content: OtpPage) -> None:
await content.code.fill(self.code)
await content.submit.click()
# endregion docs: browser-login-operations
# region docs: browser-login-router
# examples/mail/browser/login_probe.py
class LoginPortal(AsyncBrowserApi):
errors = ACCOUNT_FAILURES
identify = op(Identify)
password = op(SubmitPassword)
otp = op(SubmitOtp)
# endregion docs: browser-login-router
# region docs: browser-login-flow
# examples/mail/browser/login_probe.py
async def typed_login(portal: LoginPortal) -> Mailbox:
"""Make loss of a step type fail the ordinary mypy gate."""
password: PasswordStep = await portal.identify(email="ada@mail.example")
otp: OtpStep = await password.submit(password=VALID_LOGIN_INPUT)
mailbox: Mailbox = await otp.submit(code="123456")
return mailbox
# endregion docs: browser-login-flow
PasswordStep и OtpStep хранят тот же LoginPortal, поэтому вызывающему коду не нужно
пересобирать клиент между страницами. HTTP-линия передаёт login_id, браузерная сохраняет
контекст страницы; обе следуют одному порядку сайта.