Python SDK

pip install mailinfra

Требуется Python 3.10+. Зависимости — httpx и pydantic.

Первое письмо

from mailinfra import MailInfra

client = MailInfra()  # ключ из переменной окружения MAILINFRA_API_KEY

result = client.emails.send(
    from_email="hello@ваш-домен.ru",
    to="user@example.com",
    subject="Добро пожаловать!",
    html="<p>Привет!</p>",
    text="Привет!",
)
print(result.id, result.status)

export MAILINFRA_API_KEY=mi_live_xxxxxxxx

Ключ можно передать и явно: MailInfra(api_key="mi_live_…"). Тестовый ключ mi_test_… работает через тот же клиент: письмо сохраняется со статусом SIMULATED, не доставляется и не расходует месячный объём плана.

Защита от перепутанных ключей

MailInfra(require_environment="live") не даст клиенту создаться, если в конфигурации оказался тестовый ключ. Обратное тоже работает: require_environment="test" в тестах гарантирует, что боевой ключ не отправит настоящие письма.

Отправка по шаблону

client.emails.send(
    from_email="no-reply@ваш-домен.ru",
    to=["anna@example.com", "boris@example.com"],
    template="welcome",                     # тема берётся из шаблона
    variables={"user_name": "Анна", "app_name": "MyApp"},
    tags=["onboarding"],
)

Вложения

SDK сам кодирует байты в base64:

from mailinfra import Attachment

client.emails.send(
    from_email="billing@ваш-домен.ru",
    to="user@example.com",
    subject="Счёт за май",
    text="Счёт во вложении.",
    attachments=[
        Attachment(
            filename="invoice.pdf",
            content=open("invoice.pdf", "rb").read(),
            content_type="application/pdf",
        )
    ],
)

Идемпотентность

Ключ ставится автоматически на каждый вызов, поэтому внутренние повторы SDK безопасны. Если у операции есть естественный идентификатор — передайте его, и повтор не создаст второе письмо даже через сутки:

result = client.emails.send(
    from_email="no-reply@ваш-домен.ru",
    to="user@example.com",
    template="order_paid",
    variables={"order_id": 4242},
    idempotency_key="order_4242_paid",
)

if result.replayed:
    print("Это письмо уже было отправлено раньше")

result.idempotency_key содержит фактически использованный ключ — его стоит логировать.

Пакетная отправка

batch = client.emails.send_batch([
    {"from_email": "no-reply@ваш-домен.ru", "to": "a@example.com",
     "template": "welcome", "variables": {"user_name": "Анна"}},
    {"from_email": "no-reply@ваш-домен.ru", "to": "b@example.com",
     "template": "welcome", "variables": {"user_name": "Борис"}},
])

for item in batch.results:
    if item.error:
        print(item.index, item.error.code, item.error.message)

Батч не идемпотентен

POST /v1/emails/batch не принимает Idempotency-Key, а письма внутри батча принимаются по одному. Поэтому SDK не повторяет батч после 5xx — повтор мог бы отправить часть писем дважды. Повторяются только 429 и отказы на соединении. Ответ 202 не означает успех всего массива: проверяйте error у каждого элемента.

Чтение статусов

email = client.emails.get(result.id)
print(email.status, email.delivered_at, email.final_smtp_code)

for event in client.emails.events(result.id):
    print(event.timestamp, event.type, event.data)

bounced = client.emails.list(status="BOUNCED", tag="onboarding", limit=50)

Enum-поля — подклассы str, поэтому работают оба сравнения:

from mailinfra import EmailStatus

email.status == "DELIVERED"            # True
email.status is EmailStatus.DELIVERED  # True

Ошибки

from mailinfra import (
    DomainNotVerifiedError,
    RateLimitError,
    UnprocessableError,
    MailInfraError,
)

try:
    client.emails.send(...)
except DomainNotVerifiedError:
    ...  # домен отправителя ещё не подтверждён в DNS
except UnprocessableError as exc:
    for field in exc.fields:
        print(field["field"], field["message"])
except RateLimitError as exc:
    print("повторить через", exc.retry_after, "с")
except MailInfraError as exc:
    print(exc.code, exc.message, exc.request_id)

У любой ошибки есть .code, .message, .details, .status_code и .request_id. Последний стоит логировать: с ним поддержка найдёт запрос в своих логах. Полный список кодов — в разделе Ошибки и лимиты.

Повторы и таймауты

from mailinfra import MailInfra, RetryPolicy

client = MailInfra(timeout=10.0, max_retries=4)

client = MailInfra(
    retry_policy=RetryPolicy(
        max_retries=3,
        initial_backoff=0.5,
        backoff_factor=2.0,
        max_backoff=8.0,
        max_retry_after=60.0,  # дольше не ждём — отдаём RateLimitError
    )
)

Прочее: auto_idempotency=False отключает авто-ключ, http_client=httpx.Client(...) подставляет свой транспорт (прокси, mTLS, пул).

Асинхронный клиент

from mailinfra import AsyncMailInfra

async with AsyncMailInfra() as client:
    result = await client.emails.send(
        from_email="hello@ваш-домен.ru",
        to="user@example.com",
        subject="Привет",
        text="Привет!",
    )

Набор методов и поведение — те же.

Вебхуки

import os

from fastapi import FastAPI, Request
from mailinfra import webhooks

app = FastAPI()


@app.post("/mailinfra/events")
async def handle(request: Request) -> dict[str, str]:
    event = webhooks.construct_event(
        body=await request.body(),  # именно сырые байты
        secret=os.environ["MAILINFRA_WEBHOOK_SECRET"],
        signature_header=request.headers["X-MailInfra-Signature"],
    )

    if event.type == "email.bounced":
        mark_bounced(event.email_id, event.data["event"])

    return {"status": "ok"}

construct_event проверяет подпись и её возраст (по умолчанию 5 минут — как на стороне MailInfra) и только потом разбирает JSON. Если нужен просто флаг — есть webhooks.verify(...) -> bool.

Только сырое тело

Подпись считается от байтов запроса. Пересериализованный JSON (json.dumps после request.json()) даст другие байты, и подпись не сойдётся. Подробнее — Проверка подписи.