Pydantic v2: парсинг на границе, доверие внутри — коэрция, валидаторы и фильтрация response_model

Парсить, а не проверять

Зачем вообще разделять эти два подхода? Слой, который проверяет, оставляет бизнес-коду вопрос «а точно ли это int или строка всё-таки просочилась?» — слой, который парсит, делает этот вопрос бессмысленным. Философия pydantic в том, что граница не должна проверять данные — она должна потреблять нетипизированный вход и производить типизированный объект, либо громко падать. После успешного Order.model_validate(payload) каждый потребитель ниже по течению держит Order, чьи поля гарантированно int, str, datetime — без оборонительных isinstance, без веток «а вдруг None» в бизнес-логике. В FastAPI парсинг встроен в сигнатуру эндпоинта, а отказ автоматический:

from pydantic import BaseModel

class OrderIn(BaseModel):
    sku: str
    quantity: int
    cents: int

@app.post("/orders")
async def create_order(order: OrderIn):   # распарсено до вашего кода
    charge(order.cents * order.quantity)  # int-ы, гарантированно

Payload, который не парсится, до хендлера не доходит — клиент получает 422 с машиночитаемым списком ошибок (loc, msg, type на каждый отказ). Цена честная и небольшая: валидация pydantic v2 выполняется в pydantic-core, компилированном Rust, за наносекунды-микросекунды на поле — маленькая модель парсится за ~1–2 мкс. Заметной она становится на объёме: список из 10 тыс. вложенных моделей — это десятки миллисекунд, реальные деньги p99 на горячем эндпоинте. Сначала измеряйте, потом обвиняйте — но и не делайте вид, что граница бесплатна.

Lax против strict: где коэрция спасает, а где врёт

По умолчанию pydantic валидирует в lax-режиме по документированной таблице преобразований: строка "1" становится числом 1, "true" — True, целочисленный float 89.0 проходит в поле int. Это сознательно — веб строково типизирован (query-параметры, поля форм, переменные окружения приходят строками), и lax-режим это поглощает. Режим отказа — Хук: коэрция не просто принимает неряшливый вход, она скрывает сам факт его неряшливости. Уплывший контракт продолжает валидироваться до того дня, когда дрейф породит нечто некоэрцируемое — "89,99", — и тогда он падает в худший возможный момент вместо первого возможного. Ручка крутится на уровне поля, модели или вызова:

class PaymentIn(BaseModel):
    model_config = ConfigDict(strict=True)   # int обязан БЫТЬ int-ом
    cents: int
    currency: str

# или на вызов, на выбранных границах:
PaymentIn.model_validate(payload, strict=True)

Сеньорский паттерн — асимметричная строгость: lax там, где строки структурны (query-параметры, заголовки, настройки), strict там, где типы и есть контракт (деньги, идентификаторы, всё, что сериализует партнёр). Межполевые правила живут в валидаторах — field_validator для одного поля, model_validator(mode="after") для инвариантов между полями («возврат не может превышать исходный платёж»). И у валидаторов одно железное правило: никаких побочных эффектов. Валидатор выполняется везде, где выполняется валидация, — парсинг запроса, сериализация через response_model, вызов model_validate в батч-джобе, — так что инкремент метрики или аудит-запись внутри него срабатывает по два-три раза на запрос и по разу на каждый реплей; именно так у команд появляются аудит-таблицы, не сходящиеся с реальностью.

Граница на выходе: response_model как обратный клапан

У валидации есть зеркальное отражение на выходе. Хендлер, возвращающий свой ORM-объект, возвращает всё, что на нём есть, — так и случается классический инцидент: у User появляется колонка hashed_password, эндпоинт возвращает user, и хеш уезжает каждому клиенту, пока кто-нибудь не вчитается в JSON. response_model — это клапан: FastAPI валидирует и сериализует возвращаемое значение через объявленную модель, и поля, не объявленные на ней, в ответе просто не существуют:

class UserPublic(BaseModel):
    id: int
    email: str
    model_config = ConfigDict(from_attributes=True)  # читать из ORM-объектов

@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int):
    return await db.fetch_user(user_id)   # полная ORM-строка на входе, два поля на выходе

Фильтрация структурная, а не рекомендательная: она работает, даже когда функция возвращает более богатый объект, и продолжит работать, когда миграция следующего года добавит следующую чувствительную колонку. Та же дисциплина применима к конфигурации: pydantic-settings (библиотека для типизированного чтения переменных окружения) парсит переменные окружения через типизированную модель на старте — DATABASE_URL: PostgresDsn, TOKEN_TTL: timedelta, API_KEY: SecretStr (тип, чьё значение скрывается в repr как **********), — так что битый env роняет деплой на буте, а не первый запрос в три ночи, а секреты печатаются в логах безопасно.

Полиморфные payload-ы: размеченные объединения

Вебхуки и событийные API шлют много форм через одну дверь. Обычный Union[A, B, C, D] заставляет pydantic пробовать членов, пока один не подойдёт, — медленно, и хуже того: ошибка для плохого payload-а — невнятная куча отказов всех членов сразу. Размеченное объединение (discriminated union — объединение типов с явным полем-дискриминатором) называет поле-тег, и диспетчеризация становится одним поиском по словарю:

class PaymentSucceeded(BaseModel):
    type: Literal["payment.succeeded"]
    cents: int

class RefundCreated(BaseModel):
    type: Literal["refund.created"]
    cents: int
    original_id: str

Event = Annotated[PaymentSucceeded | RefundCreated, Field(discriminator="type")]

Неизвестный тег падает с одной точной ошибкой, указывающей на type, а не с четырёхсторонним завалом; стоимость валидации остаётся плоской при росте объединения. Для приёма событий на любом объёме дискриминатор — разница между ошибкой, по которой партнёр может действовать сам, и ошибкой, по которой он заведёт тикет.