Отправка письма без опасного повтора#

Композер отвечает одним из трёх способов: показывает подтверждение, открывает диалог об отклонённом адресе или не показывает ничего. Последний случай не означает ошибку доставки. Клиент не повторяет письмо, а ищет его в «Отправленных» по получателю и теме.

Ответ сайта	Правило HTTP или признак страницы	Что получает вызывающий
`outcome: sent`	успешный JSON или видимый `[role="status"]`	`SendResult` или `Sent`
`recipient_rejected`	типизированная ошибка или открытый `dialog`	`RecipientRejected` или `Rejected`
`outcome: silent`	JSON без подтверждения на странице	`silent`, затем результат поиска, или `Silent`

→ Request · POST /api/v1/messages/send

Authorization: Bearer session-1
Content-Type: application/json
X-Mail-Signature: 4a0a5ef967590a6eeec04612bc5a0066c0826b0c3d36af223ab541bc2055c815

{"recipient":"grace@mail.example","subject":"Проверка доставки","body":"Встречаемся в пятницу."}

← Response · 200 OK

{
  "status": "ok",
  "body": {"outcome": "sent", "message_id": 100}
}

HTTP

Подпись строится после сериализации тела. body_digest("sha256") даёт hex-строку хеша, hmac_sha256(...) подписывает её учебным ключом, а header_output(...) записывает результат в X-Mail-Signature.

# examples/mail/http/send.py
MAIL_BEARER = BearerScheme("mail-bearer")
MAIL_KEY = SigningKeyRequirement("mail-send")
MAIL_HMAC = hmac_sha256(
    key=MAIL_KEY,
    base=body_digest("sha256"),
    output=header_output("X-Mail-Signature"),
)


def signing_key(requirement: SigningKeyRequirement) -> SigningKey:
    if requirement == MAIL_KEY:
        return SigningKey(b"mail-demo-secret")
    raise LookupError(requirement.name)

SendMessage отличает успешный конверт от отклонённого адреса по Const. Поиск в «Отправленных» остаётся отдельной идемпотентной GET-операцией.

# examples/mail/http/send.py
@dataclass(frozen=True, slots=True, kw_only=True)
class SendMessage(HttpOperation[SendResult]):
    __http__ = Http.post(
        "/api/v1/messages/send",
        success={200: Json(SendEnvelope)},
        errors={200: (Json(RejectedEnvelope), RecipientRejected)},
        security=MAIL_BEARER,
        signing=MAIL_HMAC,
    )

    body: Annotated[NewMessage, JsonBody()]


@dataclass(frozen=True, slots=True, kw_only=True)
class FindSent(HttpOperation[SentMessages]):
    __http__ = Http.get(
        "/api/v1/messages/sent",
        success={200: Json(SentMessagesEnvelope)},
        security=MAIL_BEARER,
    )

    recipient: Query[str]
    subject: Query[str]


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

    send = op(SendMessage)
    sent = op(FindSent)

После silent код не вызывает send второй раз. Он ищет письмо с теми же получателем и темой; найденный идентификатор 101 доказывает, что первый запрос уже сработал.

# examples/mail/http/send.py
def send_three_messages() -> tuple[int, str, str, int]:
    identity = Identity(
        auth=(MAIL_BEARER.static("session-1"),),
        key_provider=signing_key,
    )
    with mail_client() as client:
        api = MailApi(client, identity=identity)
        sent = api.send(
            body=NewMessage(
                recipient="grace@mail.example",
                subject="Проверка доставки",
                body="Встречаемся в пятницу.",
            )
        )
        try:
            api.send(
                body=NewMessage(
                    recipient="rejected@mail.example",
                    subject="Проверка адреса",
                    body="Это письмо отклонят.",
                )
            )
        except RecipientRejected as error:
            rejected = type(error).__name__

        silent = api.send(
            body=NewMessage(
                recipient="quiet@mail.example",
                subject="Тихая доставка",
                body="Проверим папку отправленных.",
            )
        )
        found = api.sent(
            recipient="quiet@mail.example",
            subject="Тихая доставка",
        )

    return sent.message_id or 0, rejected, silent.outcome, found.items[0].id

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

"""Sign a mail send request and handle sent, rejected and silent outcomes."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated

from pydantic import BaseModel

from eazy_sdk import Http, HttpOperation, Identity, Query, SyncApi, op
from eazy_sdk.auth import BearerScheme
from eazy_sdk.request import (
    SigningKey,
    SigningKeyRequirement,
    body_digest,
    header_output,
    hmac_sha256,
)
from eazy_sdk.request.markers import JsonBody
from eazy_sdk.response import ApiError, Const, Json, Payload

from examples.mail.http._client import mail_client


class NewMessage(BaseModel):
    recipient: str
    subject: str
    body: str


class SendResult(BaseModel):
    outcome: str
    message_id: int | None = None


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


class RejectedDetails(BaseModel):
    code: Annotated[str, Const("recipient_rejected")]
    message: str


class RejectedEnvelope(BaseModel):
    status: Annotated[str, Const("recipient_rejected")]
    body: RejectedDetails


class RecipientRejected(ApiError[RejectedEnvelope]):
    pass


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


class SentMessages(BaseModel):
    items: list[SentMessage]


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


# region docs: http-send-signature
# examples/mail/http/send.py
MAIL_BEARER = BearerScheme("mail-bearer")
MAIL_KEY = SigningKeyRequirement("mail-send")
MAIL_HMAC = hmac_sha256(
    key=MAIL_KEY,
    base=body_digest("sha256"),
    output=header_output("X-Mail-Signature"),
)


def signing_key(requirement: SigningKeyRequirement) -> SigningKey:
    if requirement == MAIL_KEY:
        return SigningKey(b"mail-demo-secret")
    raise LookupError(requirement.name)
# endregion docs: http-send-signature


# region docs: http-send-operations
# examples/mail/http/send.py
@dataclass(frozen=True, slots=True, kw_only=True)
class SendMessage(HttpOperation[SendResult]):
    __http__ = Http.post(
        "/api/v1/messages/send",
        success={200: Json(SendEnvelope)},
        errors={200: (Json(RejectedEnvelope), RecipientRejected)},
        security=MAIL_BEARER,
        signing=MAIL_HMAC,
    )

    body: Annotated[NewMessage, JsonBody()]


@dataclass(frozen=True, slots=True, kw_only=True)
class FindSent(HttpOperation[SentMessages]):
    __http__ = Http.get(
        "/api/v1/messages/sent",
        success={200: Json(SentMessagesEnvelope)},
        security=MAIL_BEARER,
    )

    recipient: Query[str]
    subject: Query[str]


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

    send = op(SendMessage)
    sent = op(FindSent)
# endregion docs: http-send-operations


# region docs: http-send-flow
# examples/mail/http/send.py
def send_three_messages() -> tuple[int, str, str, int]:
    identity = Identity(
        auth=(MAIL_BEARER.static("session-1"),),
        key_provider=signing_key,
    )
    with mail_client() as client:
        api = MailApi(client, identity=identity)
        sent = api.send(
            body=NewMessage(
                recipient="grace@mail.example",
                subject="Проверка доставки",
                body="Встречаемся в пятницу.",
            )
        )
        try:
            api.send(
                body=NewMessage(
                    recipient="rejected@mail.example",
                    subject="Проверка адреса",
                    body="Это письмо отклонят.",
                )
            )
        except RecipientRejected as error:
            rejected = type(error).__name__

        silent = api.send(
            body=NewMessage(
                recipient="quiet@mail.example",
                subject="Тихая доставка",
                body="Проверим папку отправленных.",
            )
        )
        found = api.sent(
            recipient="quiet@mail.example",
            subject="Тихая доставка",
        )

    return sent.message_id or 0, rejected, silent.outcome, found.items[0].id
# endregion docs: http-send-flow


def main() -> None:
    sent, rejected, silent, found = send_three_messages()
    print(f"sent: {sent}")
    print(f"rejected: {rejected}")
    print(f"silent response: {silent}")
    print(f"found in Sent: {found}")


if __name__ == "__main__":
    main()

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

sent: 100
rejected: RecipientRejected
silent response: silent
found in Sent: 101

Браузер

Страница /compose/ содержит форму и два наблюдаемых признака. Третий исход определяется по истечению общего срока ожидания, когда ни подтверждение, ни диалог не появились.

<main data-page="compose">
  <form data-composer>
    <input name="recipient" type="email"><input name="subject">
    <textarea name="body"></textarea><button type="submit">Отправить</button>
  </form>
  <div role="status" hidden>Отправлено</div>
  <dialog data-outcome="rejected">Адрес отклонён</dialog>
</main>

HMAC здесь вычисляет JavaScript самой учебной страницы перед fetch. Браузерная операция отвечает только за поля формы, клик и распознавание результата через outcomes(...).

# examples/mail/browser/send.py
class ComposerPage:
    recipient: Annotated[Element, css('input[name="recipient"]')]
    subject: Annotated[Element, css('input[name="subject"]')]
    body: Annotated[Element, css('textarea[name="body"]')]
    submit: Annotated[Element, css('button[type="submit"]')]
    rejected: Annotated[Element, css('[data-outcome="rejected"]')]


@dataclass(frozen=True, slots=True)
class Sent:
    pass


@dataclass(frozen=True, slots=True)
class Rejected:
    pass


@dataclass(frozen=True, slots=True)
class Silent:
    pass


async def _sent(content: ComposerPage) -> Sent:
    _ = content
    return Sent()


async def _rejected(content: ComposerPage) -> Rejected:
    _ = content
    return Rejected()


async def _silent(content: ComposerPage) -> Silent:
    _ = content
    return Silent()


@dataclass(frozen=True, slots=True, kw_only=True)
class OpenComposer(BrowserOperation[None, None]):
    __browser__ = Browser.goto("/compose/", at=css("[data-composer]"))


@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitMessage(BrowserOperation[ComposerPage, Sent | Rejected | Silent]):
    __browser__ = Browser.act(
        ComposerPage,
        at=css("[data-composer]"),
        outcomes=outcomes(
            when(visible(css('[role="status"]')), then=_sent),
            when(visible(css('[data-outcome="rejected"]')), then=_rejected),
            otherwise=_silent,
        ),
    )

    recipient: str
    subject: str
    body: str

    async def act(self, content: ComposerPage) -> None:
        if await content.rejected.visible():
            await content.rejected.press("Escape")
        await content.recipient.fill(self.recipient)
        await content.subject.fill(self.subject)
        await content.body.fill(self.body)
        await content.submit.click()


class ComposerApi(AsyncBrowserApi):
    open = op(OpenComposer)
    send = op(SubmitMessage)

Три последовательных вызова идут через настоящий Chromium и один MailSite. Перед новым вызовом операция закрывает диалог предыдущего отказа, если он ещё открыт.

# examples/mail/browser/send.py
async def send_three_messages() -> tuple[
    Sent | Rejected | Silent,
    Sent | Rejected | Silent,
    Sent | Rejected | Silent,
]:
    async with playwright_mail() as runtime:
        composer = ComposerApi(runtime.client)
        await composer.open(options=OPTIONS)
        sent = await composer.send(
            recipient="grace@mail.example",
            subject="Проверка доставки",
            body="Встречаемся в пятницу.",
            options=OPTIONS,
        )
        rejected = await composer.send(
            recipient="rejected@mail.example",
            subject="Проверка адреса",
            body="Это письмо отклонят.",
            options=OPTIONS,
        )
        silent = await composer.send(
            recipient="quiet@mail.example",
            subject="Тихая доставка",
            body="Проверим папку отправленных.",
            options=OPTIONS,
        )
        return sent, rejected, silent

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

"""Submit the mail composer and distinguish its three visible outcomes."""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from typing import Annotated

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

from examples.mail.site._playwright import playwright_mail

OPTIONS = BrowserCallOptions(timeout=1.0, element_timeout=5.0)


# region docs: browser-send-operation
# examples/mail/browser/send.py
class ComposerPage:
    recipient: Annotated[Element, css('input[name="recipient"]')]
    subject: Annotated[Element, css('input[name="subject"]')]
    body: Annotated[Element, css('textarea[name="body"]')]
    submit: Annotated[Element, css('button[type="submit"]')]
    rejected: Annotated[Element, css('[data-outcome="rejected"]')]


@dataclass(frozen=True, slots=True)
class Sent:
    pass


@dataclass(frozen=True, slots=True)
class Rejected:
    pass


@dataclass(frozen=True, slots=True)
class Silent:
    pass


async def _sent(content: ComposerPage) -> Sent:
    _ = content
    return Sent()


async def _rejected(content: ComposerPage) -> Rejected:
    _ = content
    return Rejected()


async def _silent(content: ComposerPage) -> Silent:
    _ = content
    return Silent()


@dataclass(frozen=True, slots=True, kw_only=True)
class OpenComposer(BrowserOperation[None, None]):
    __browser__ = Browser.goto("/compose/", at=css("[data-composer]"))


@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitMessage(BrowserOperation[ComposerPage, Sent | Rejected | Silent]):
    __browser__ = Browser.act(
        ComposerPage,
        at=css("[data-composer]"),
        outcomes=outcomes(
            when(visible(css('[role="status"]')), then=_sent),
            when(visible(css('[data-outcome="rejected"]')), then=_rejected),
            otherwise=_silent,
        ),
    )

    recipient: str
    subject: str
    body: str

    async def act(self, content: ComposerPage) -> None:
        if await content.rejected.visible():
            await content.rejected.press("Escape")
        await content.recipient.fill(self.recipient)
        await content.subject.fill(self.subject)
        await content.body.fill(self.body)
        await content.submit.click()


class ComposerApi(AsyncBrowserApi):
    open = op(OpenComposer)
    send = op(SubmitMessage)
# endregion docs: browser-send-operation


# region docs: browser-send-flow
# examples/mail/browser/send.py
async def send_three_messages() -> tuple[
    Sent | Rejected | Silent,
    Sent | Rejected | Silent,
    Sent | Rejected | Silent,
]:
    async with playwright_mail() as runtime:
        composer = ComposerApi(runtime.client)
        await composer.open(options=OPTIONS)
        sent = await composer.send(
            recipient="grace@mail.example",
            subject="Проверка доставки",
            body="Встречаемся в пятницу.",
            options=OPTIONS,
        )
        rejected = await composer.send(
            recipient="rejected@mail.example",
            subject="Проверка адреса",
            body="Это письмо отклонят.",
            options=OPTIONS,
        )
        silent = await composer.send(
            recipient="quiet@mail.example",
            subject="Тихая доставка",
            body="Проверим папку отправленных.",
            options=OPTIONS,
        )
        return sent, rejected, silent
# endregion docs: browser-send-flow


async def run() -> None:
    sent, rejected, silent = await send_three_messages()
    print(f"sent: {type(sent).__name__}")
    print(f"rejected: {type(rejected).__name__}")
    print(f"silent: {type(silent).__name__}")


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


if __name__ == "__main__":
    main()

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

sent: Sent
rejected: Rejected
silent: Silent

Предупреждение

Не повторяйте отправку после тишины в интерфейсе: запрос мог завершиться успешно. Сначала найдите письмо в «Отправленных» по устойчивым полям.

Отправка письма без опасного повтора#

Дальше#