Webhook в FastAPI: выбираем модель события по полю type

Платёжная система присылает в один адрес несколько событий. Для успешной оплаты нужны номер заказа и сумма, для возврата — номер исходного платежа и причина. Если сложить всё в одну модель с необязательными полями, endpoint примет комбинации, которые отправитель никогда не должен создавать. Проверки начнут расползаться по веткам обработчика.

Пусть вид события выбирает форму данных

У каждого сообщения есть поле type. Используем его как явный переключатель: payment выбирает модель Payment, refund — Refund. Значение Literal в каждой модели фиксирует разрешённую метку. Аннотация объединения с Field(discriminator="type") сообщает Pydantic, какое поле прочитать перед проверкой остальных данных.

Это отличается от обычного объединения типов. Серверу не приходится угадывать вариант по тому, какие поля случайно подошли. Ошибка относится к выбранному событию: возврат без payment_id остаётся некорректным возвратом. Добавление следующего варианта не меняет смысл уже существующих меток, если контракт отправителя сохраняется.

Сначала перечислите события, которые приложение действительно умеет обрабатывать. Универсальная ветка со словарём принимает новые сообщения незаметно, но переносит проблему в бизнес-логику. Для интеграции с обязательной поддержкой неизвестных событий нужен отдельный явный сценарий: например, сохранить исходное сообщение и отправить его на разбор. В нашем endpoint неизвестный тип получает 422.

Проверим оба события через HTTP

Нужны FastAPI, HTTPX и Pydantic v2. Один полный блок создаёт приложение и выполняет запросы локально через TestClient. В ответе используются разные поля вариантов: это показывает, что маршрут получил подходящий объект, а не только словарь с правильной строкой type.

Запрет лишних полей здесь помогает заметить смешанное событие: payment с полями refund тоже должно считаться ошибкой. Он дополняет выбор модели, но не заменяет discriminator. Уберите переключатель — и логика выбора снова будет зависеть от проверки объединения целиком.

from typing import Annotated, Literal
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field

class Payment(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["payment"]
    order_id: int
    amount_cents: int = Field(gt=0)

class Refund(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["refund"]
    payment_id: int
    reason: str = Field(min_length=1)

Event = Annotated[Payment | Refund, Field(discriminator="type")]
app = FastAPI()
@app.post("/webhook")
def webhook(event: Event):
    if isinstance(event, Payment):
        return {"action": "paid", "id": event.order_id}
    return {"action": "refund", "id": event.payment_id}

with TestClient(app) as client:
    paid = client.post("/webhook", json={"type": "payment", "order_id": 7, "amount_cents": 150})
    refund = client.post("/webhook", json={"type": "refund", "payment_id": 9, "reason": "return"})
    assert paid.json() == {"action": "paid", "id": 7}
    assert refund.json() == {"action": "refund", "id": 9}
    unknown = client.post("/webhook", json={"type": "cancel"})
    assert unknown.status_code == 422
    assert unknown.json()["detail"][0]["type"] == "union_tag_invalid"
    missing = client.post("/webhook", json={"type": "refund", "reason": "return"})
    assert missing.status_code == 422
    assert missing.json()["detail"][0]["loc"][-1] == "payment_id"
print(paid.json(), refund.json(), unknown.status_code)

Где заканчивается проверка формы

Пример возвращает два действия с разными идентификаторами и отклоняет неизвестную метку. Добавьте к нему проверку OpenAPI: схема тела должна содержать варианты и discriminator. Генератор клиента тогда видит тот же переключатель, которым пользуется сервер. Если в схеме внезапно остался один объект с десятком optional-полей, проверьте аннотацию параметра и не потерялся ли Annotated при переносе типов.

Проверка модели ещё не подтверждает подпись webhook. Перед изменением заказа нужно проверить происхождение сообщения, найти заказ и сверить допустимое состояние. Положительное число amount_cents описывает форму, но ничего не говорит о сумме именно этого заказа. Эти проверки выполняются после разбора выбранного события и используют уже известные поля конкретного варианта.

Отдельно продумайте повторную доставку. Один payment может прийти несколько раз, а корректная модель примет каждый повтор. Используйте идентификатор события отправителя и устойчивое хранилище результата обработки. Связывать дедупликацию только с order_id опасно: один заказ способен иметь несколько разных событий. В рассматриваемом коде записи не выполняются, поэтому он проверяет только входной договор и выбор ветки.

При добавлении новой метки начните с нового класса и HTTP-теста реального сообщения. Затем обновите объединение и обработчик. Избегайте переименования существующего значения type ради красоты Python-кода: это уже внешний контракт. Внутреннее имя класса можно менять отдельно, сохранив Literal и обязательные поля.

Для диагностики сохраняйте метку и идентификатор события рядом с причиной отказа. Полное тело платежного сообщения может содержать персональные данные. Чёткая схема облегчает и журналирование: на раннем этапе видно, отказал ли выбор варианта, обязательное поле или уже бизнес-проверка. Эти причины требуют разных действий от владельца интеграции.

Следующая практика

Перед несколькими вариантами событий полезно закрепить одну JSON-модель: обязательные поля, ограничения и ошибочный запрос. Начните с открытой практики POST, затем расширьте её объединением из статьи. Откройте практику POST и JSON-модели. Перейдите к бесплатному курсу FastAPI.