Перейти к основному содержимому

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",
        )
    ],
)

Картинка внутри письма — Attachment(..., content_id="logo"), в HTML на неё ссылаются как <img src="cid:logo">.

Metadata, отложенная отправка и срок жизни​

from datetime import datetime, timedelta, timezone

reminder = client.emails.send(
    from_email="no-reply@ваш-домен.ru",
    to="user@example.com",
    template="webinar_reminder",
    metadata={"webinar_id": "w-17"},  # вернётся в письме и вебхуках
    send_at=datetime.now(timezone.utc) + timedelta(hours=23),  # только с часовым поясом
    ttl=1800,  # секунд; позже письмо уже не нужно
)
print(reminder.status)  # SCHEDULED

client.emails.reschedule(reminder.id, send_at="2026-11-01T08:00:00+03:00")
client.emails.cancel(reminder.id)  # статус CANCELED, резерв квоты вернётся
  • metadata — до 20 ключей (до 40 символов из латиницы, цифр и _.-), значения — строки до 500 символов, всё — до 4 КБ; SDK проверяет это до запроса.
  • send_at — не дальше 7 дней от приёма; дальше — SendAtTooFarError. Уже ушедшее или отменённое письмо — EmailNotScheduledError.
  • ttl — от 60 до 86 400 секунд, отсчёт от send_at или от приёма.

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

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

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)

Ответ 202 не означает успех всего массива: проверяйте error у каждого элемента.

Батч уходит с Idempotency-Key (автоматическим или idempotency_key="import-2026-10-28", до 252 символов), который переиспользуется в повторах: уже принятые письма повтор не дублирует, поэтому SDK повторяет батч и после 5xx и таймаутов. batch.replayed — ничего нового не принято. С auto_idempotency=False батч повторяется только на 429.

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

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)

page = client.emails.list(status="BOUNCED", tag="onboarding", limit=50)
print(len(page.data), page.has_more, page.next_cursor)

for email in client.emails.list_all(status="BOUNCED", limit=100):
    print(email.id, email.metadata)

list() возвращает одну страницу (до 0.2.0 — список), следующая — тот же вызов с cursor=page.next_cursor. list_all() обходит страницы сам и лениво; у AsyncMailInfra это асинхронный итератор (async for).

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. Последний стоит логировать: с ним поддержка найдёт запрос в своих логах. Конкретный класс всегда наследует класс своего HTTP-статуса: DomainNotVerifiedError — это и UnprocessableError (422). Полный список кодов — в разделе Ошибки и лимиты.

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

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
    )
)

timeout принимает и секунды, и httpx.Timeout(connect=…, read=…) для раздельных фаз. Если передан свой http_client, а timeout не задан — SDK берёт настройку вашего клиента, а не навязывает свои 30 секунд.

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

Без ключа идемпотентности повторов нет

Ретраить 5xx и таймауты у send() безопасно ровно потому, что SDK ставит Idempotency-Key и переиспользует его на повторах. С auto_idempotency=False этой гарантии нет, поэтому SDK повторяет такую отправку только на 429 — как батч.

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

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. Во время ротации секрета в заголовке две подписи v1 — SDK принимает любую из них. event.metadata — metadata письма из события.

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

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