Валидация схемы в Python: как течёт данные в живом сервисе

Сервис принимал вебхуки от платёжного провайдера, складывал их в очередь и раз в сутки собирал отчёт по выручке. Раз в неделю отчёт падал с TypeError: unsupported operand type(s) for +: 'int' and 'str'. Провайдер иногда присылал поле amount строкой — "1990" вместо 1990. Ручная проверка на входе ловила отсутствие ключа, но не тип. Разбираем на этом кейсе, где именно протекает валидация в Python-сервисе, что даёт переход на схему и чем платишь за строгость под нагрузкой.

Версии в разборе: Python 3.12, pydantic 2.x. Между pydantic 1.x и 2.x поведение и API поменялись радикально — если у вас 1.x, часть примеров не заработает, об этом отдельно ниже.

Где течёт: граница системы

Валидацию имеет смысл ставить ровно там, где данные пересекают границу вашего контроля: HTTP-запрос, тело вебхука, строка из очереди, ответ внешнего API, содержимое конфиг-файла. Внутри процесса данные уже ваши — там валидация превращается в дублирование.

В исходном коде проверка выглядела так:

def handle_webhook(payload: dict) -> None:
    # проверяем, что ключи на месте
    if "amount" not in payload or "currency" not in payload:
        raise ValueError("missing fields")
    save_to_queue(payload)  # payload как пришёл, без нормализации

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

Валидация без нормализации — половина работы. Проверить, что ключ есть, и не привести его к нужному типу — значит отложить падение на потом и в другое место.

Что даёт схема

Схема на pydantic делает три вещи за один вызов: проверяет наличие полей, проверяет и приводит типы, отдаёт объект с гарантированными атрибутами. После успешной валидации остальной код может складывать amount без оглядки.

from decimal import Decimal
from pydantic import BaseModel, field_validator

class PaymentEvent(BaseModel):
    amount: Decimal          # деньги — только Decimal, не float
    currency: str
    external_id: str

    @field_validator("currency")
    @classmethod
    def currency_upper(cls, v: str) -> str:
        # нормализуем регистр: rub и RUB — одно и то же
        return v.upper()


def handle_webhook(payload: dict) -> None:
    event = PaymentEvent.model_validate(payload)  # тут упадёт, если тип не тот
    save_to_queue(event.model_dump(mode="json"))

Строку "1990" pydantic по умолчанию приведёт к Decimal("1990"). Отчёт больше не встретит строку. А если провайдер пришлёт amount: "abc" — падение произойдёт на строке model_validate, в момент приёма вебхука, с понятным сообщением, а не через сутки в агрегаторе.

Строгость приведения: главная развилка

pydantic 2.x по умолчанию работает в lax-режиме: "1990" станет числом, 1 станет True, "2024-01-01" станет датой. Это удобно для вебхуков, где источник сам плавает по типам. Но то же приведение опасно, если источник — ваш собственный внутренний сервис, где строка вместо числа означает баг, а не вариант нормы.

Для строгого режима есть strict=True на уровне поля или всей модели:

from pydantic import BaseModel, ConfigDict

class InternalCommand(BaseModel):
    model_config = ConfigDict(strict=True)
    # теперь "1990" для int-поля вызовет ошибку, а не приведение
    retries: int

Точное поведение приведения по каждому типу описано в разделе Conversion Table документации pydantic — оно менялось между минорными версиями 2.x, так что сверяйтесь с таблицей для вашей версии, а не запоминайте.

Подходы, между которыми выбирают

Важное про @dataclass и аннотации типов вообще: аннотация amount: int в обычном классе или dataclass в рантайме не проверяется. Питон не мешает положить туда строку. Если вы полагались на аннотации как на валидацию — это была иллюзия. Проверку даёт либо явный код, либо библиотека вроде pydantic, которая читает эти аннотации и применяет их.

Что мерить под нагрузкой

Схема — это не бесплатно. На каждый входящий объект создаётся экземпляр модели, прогоняются валидаторы. На потоке в сотни событий в секунду это заметно в профиле. Что имеет смысл замерить на своём железе и своих данных, прежде чем решать:

• Время валидации на объект. Постройте одну модель, прогоните типовой payload в цикле, посмотрите медиану и хвост. pydantic 2.x собран на Rust-ядре и обычно быстрее ручного дерева if на вложенных структурах — но проверьте на своей форме данных, а не поверьте на слово.
• Стоимость model_dump. Сериализация обратно в dict/JSON тоже расходует время. Если вы валидируете только чтобы тут же выгрузить в очередь, посчитайте, не дешевле ли хранить исходный dict.
• Аллокации. Тысячи короткоживущих моделей в секунду — это давление на сборщик мусора. Смотрите на паузы GC, если сервис latency-чувствительный.

Придумывать здесь конкретные миллисекунды бессмысленно — они зависят от формы объекта, глубины вложенности и версии. Меряйте timeit или профилировщиком на репрезентативном payload.

Где подход ломается

Схема решает проблему типов на границе, но создаёт свои граничные случаи.

• Лишние поля молча теряются. По умолчанию pydantic игнорирует ключи, которых нет в модели. Провайдер добавил поле fee, вы про него не знали — и не узнаете, пока не понадобится. Хотите падать на незнакомых полях — ставьте model_config = ConfigDict(extra="forbid").
• Приведение маскирует баг источника. lax-режим превратит "1990" в число и скроет тот факт, что провайдер шлёт деньги строкой. Иногда это то, что нужно; иногда — сигнал, что контракт нарушен и стоит написать провайдеру, а не молча чинить у себя.
• float для денег. Не типизируйте суммы как float. 0.1 + 0.2 в Python не равно 0.3. Для денег — Decimal, и следите, чтобы источник не отдавал их уже испорченным float.
• Вложенные необязательные структуры. Optional[SubModel] = None с приходящим null ведёт себя не так, как отсутствующий ключ. Проверьте оба случая на своих данных.

Если у вас pydantic 1.x

Разница не косметическая. В 1.x методы назывались .parse_obj() и .dict(), а не .model_validate() и .model_dump(); декоратор был @validator, а не @field_validator; конфиг задавался вложенным классом Config, а не model_config. Код из этой статьи на 1.x не запустится. Официальный гайд по миграции 1.x → 2.x есть в документации проекта — если сервис живой и на 1.x, планируйте переход отдельной задачей, а не по ходу.

Что сделать дальше

Первый практический шаг — найти в своём коде все места, где вы принимаете dict из внешнего источника и обращаетесь к его ключам напрямую. Каждое такое место — потенциальная протечка типа, которая всплывёт не здесь. Заведите модель на самую нестабильную границу (обычно это вебхуки и ответы сторонних API), включите на ней extra="forbid" на время отладки, чтобы увидеть, какие поля источник реально шлёт, и решите по каждому: приводить в lax или падать в strict. После этого замерьте валидацию под своей реальной нагрузкой, прежде чем раскатывать на все границы.