← назад к разделу

Встроенных ограничений Pydantic — Field(min_length=12), Field(ge=1), EmailStr, UUID — хватает для простых случаев. Когда логика сложнее, пишут свои правила. Разберём, как это сделать, как проверить согласованность нескольких полей и как превратить английские msg из ValidationError в понятные клиенту тексты: переводов у Pydantic нет вовсе.

Три типичных правила показывают всю развилку сразу: одно станет типом поля, второе — проверкой на уровне модели, а третьему в валидаторе вообще не место.

Обязательно

Зачем нужны свои правила

Встроенные ограничения проверяют одно поле по простому критерию. Они не умеют:

  • сверять значение с базой («такой логин уже занят»);
  • проверять бизнес-правило, специфичное для домена («скидка не может быть больше цены»);
  • давать одно имя составному правилу, которое иначе копируется Field(pattern=...) по десяти моделям.

Первое — не для валидатора вообще. Остальные два — свои правила.

Как устроено своё правило

Телефон должен начинаться с +7 и содержать одиннадцать цифр. Field(pattern=r"^\+7\d{10}$") это умеет, но выражение придётся копировать в каждую модель с телефоном, а сообщение останется английским String should match pattern. Своё правило убирает копирование: правило, его имя и сообщение живут в одном месте — в типе.

import re
from typing import Annotated

from pydantic import AfterValidator, BaseModel
from pydantic_core import PydanticCustomError

PHONE = re.compile(r"\+7\d{10}")


def check_phone(value: str) -> str:
    if not PHONE.fullmatch(value):
        raise PydanticCustomError("phone", "номер должен начинаться с +7 и содержать 11 цифр")
    return value


Phone = Annotated[str, AfterValidator(check_phone)]

class CreateUser(BaseModel):
    name: str = Field(min_length=1)
    phone: Phone
    backup_phone: Phone | None = None

Три вещи, которые здесь важны. Функция правила получает уже разобранное значение (AfterValidator работает после проверки типа) и либо возвращает его, либо бросает PydanticCustomError с именем правила и текстом: имя попадёт в поле type ошибки, по нему потом найдётся перевод. Регулярное выражение компилируется один раз на модуль, а не внутри функции: правило вызывается на каждое поле каждого запроса. И пустое значение — не забота правила: Phone | None = None означает необязательное поле, и для None правило не вызывается, а за «обязательно» отвечает отсутствие значения по умолчанию.

Короткая формула: тип описывает контракт, функция — реализует его. Правило, нужное одной модели, пишут короче — методом с @field_validator("phone"); правило, нужное десяти, — типом.

Межполевая валидация

Проверить, что два поля согласованы между собой, нельзя внутри одного поля. Для самого частого случая — «второе поле равно первому» — хватает валидатора поля, который заглядывает в уже проверенные:

from pydantic import ConfigDict, Field, ValidationInfo, field_validator
from pydantic.alias_generators import to_camel


class ChangePassword(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    password: str = Field(min_length=12)
    confirm_password: str

    @field_validator("confirm_password")
    @classmethod
    def same_as_password(cls, value: str, info: ValidationInfo) -> str:
        if "password" in info.data and value != info.data["password"]:
            raise PydanticCustomError("password_mismatch", "пароли не совпадают")
        return value

info.data содержит поля, которые объявлены раньше и уже прошли проверку. Ошибка при этом привязана к confirmPassword — ровно к тому полю, которое клиенту надо подсветить. Проверка "password" in info.data обязательна: если пароль короче двенадцати символов, в info.data его нет, и без проверки правило упадёт с KeyError вместо того, чтобы промолчать и оставить работу полевому ограничению.

Когда правило сложнее равенства — «дата окончания после начала, но не дальше чем на год», «скидка не больше цены» — пишут проверку на уровне модели:

from pydantic import model_validator


class DiscountRequest(BaseModel):
    price: int = Field(gt=0)
    discount: int = Field(ge=0)

    @model_validator(mode="after")
    def discount_not_above_price(self) -> "DiscountRequest":
        if self.discount > self.price:
            raise PydanticCustomError("lte_price", "скидка не больше цены", {"field": "discount"})
        return self

У ошибки модели нет поля: в ValidationError её loc пустой, а FastAPI покажет ("body",). Поэтому поле, к которому клиент должен привязать сообщение, передают третьим аргументом — в ctx; сборщик ответа ниже его оттуда прочитает.

Порядок: почему второй ошибки не видно

Валидатор модели с mode="after" выполняется только если все поля прошли проверку. Пользователь прислал пустую цену и скидку больше нуля — получит одно сообщение «цена обязательна», а не два: сравнивать было нечего, и Pydantic до сравнения не дошёл. Это удобно, но имеет оборотную сторону: пока в теле есть хоть одна полевая ошибка, межполевые правила молчат, и клиент узнаёт о них со второй попытки.

Если нужно увидеть всё сразу, проверку делят на два вызова: сначала модель только с полевыми ограничениями, затем, при успехе, правила согласованности. Групп и последовательностей, как в Bean Validation, у Pydantic нет, и для двух-трёх проверок они и не нужны.

Сообщения об ошибках

Главное отличие от Java-стека: Pydantic возвращает список нарушений с машинным типом, а не локализованные тексты. ValidationError.errors() даёт на каждое нарушение type (string_too_short, missing, phone), loc, английский msg и ctx с параметрами правила:

[
  {"type": "string_too_short", "loc": ("password",), "msg": "String should have at least 12 characters", "ctx": {"min_length": 12}},
  {"type": "phone", "loc": ("phone",), "msg": "номер должен начинаться с +7 и содержать 11 цифр"},
]

Переводов встроенных типов в комплекте нет: русский текст для string_too_short пишут сами. Это и есть аналог файла сообщений — словарь по type с подстановкой из ctx:

MESSAGES = {
    "missing": "{field}: обязательное поле",
    "string_too_short": "{field}: не короче {min_length} символов",
    "greater_than_equal": "{field}: не меньше {ge}",
    "extra_forbidden": "{field}: неизвестное поле",
    "json_invalid": "тело запроса не является JSON",
}


def message(error: dict, messages: dict[str, str]) -> str:
    field = ".".join(str(part) for part in error["loc"]) or None
    template = messages.get(error["type"])
    if template is None:
        return error["msg"]
    return template.format(field=field, **error.get("ctx", {}))

Своим правилам перевод не нужен: PydanticCustomError("phone", "...") уже несёт текст на нужном языке. Правило: каждый type, который вы завели сами, получает текст в месте объявления, а для встроенных типов словарь заполняют по мере появления в ответах.

Имена полей из JSON, а не из Python

alias_generator=to_camel переводит confirm_password в confirmPassword при разборе, и loc в ошибках уже содержит псевдоним: Pydantic по умолчанию сообщает имя так, как оно пришло в запросе. Отдельной настройки, откуда брать имя поля, не нужно. Нужно другое — populate_by_name=True, чтобы ту же модель можно было собрать в коде по именам Python, в тестах и фабриках.

Во что нарушение превращается в ответе

Статья про сообщения без ответа с сообщением неполна. Вот что получает клиент, когда сработали и полевое, и межполевое правило:

field_validator или model_validator raise ValueError внутри правила ValidationError с loc и msg FastAPI собирает все ошибки обработчик RequestValidationError один формат на всё API 400 с violations: field, code, message

Своё правило бросает ValueError, Pydantic собирает нарушения с путём к полю, а единый обработчик переводит их в список violations ответа.

{
  "type": "https://api.example.com/problems/validation-failed",
  "title": "Проверка не пройдена",
  "status": 400,
  "code": "VALIDATION_FAILED",
  "violations": [
    { "field": "phone",    "message": "номер должен начинаться с +7 и содержать 11 цифр", "rule": "phone" },
    { "field": "discount", "message": "скидка не больше цены",                            "rule": "lte_price" }
  ]
}

Сборка списка — одна функция на приложение, а не на обработчик:

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError


def violations(exc: RequestValidationError, messages: dict[str, str]) -> list[dict]:
    out = []
    for error in exc.errors():
        loc = [str(part) for part in error["loc"][1:]]
        field = ".".join(loc) or error.get("ctx", {}).get("field")
        if error["type"] == "json_invalid":
            field = None
        out.append({"field": field, "message": message({**error, "loc": loc}, messages), "rule": error["type"]})
    return out


@app.exception_handler(RequestValidationError)
async def on_validation_error(request: Request, exc: RequestValidationError):
    return problem(request, status=400, code="VALIDATION_FAILED", violations=violations(exc, messages_for(request)))

Первый элемент loc у FastAPI — откуда пришло значение: body, query, path, header; его отбрасывают, остальное склеивают через точку (items.0.qty). Два момента, которые решают за вас удобство клиента. У межполевого нарушения тоже есть поле — из ctx, как договорились выше. И имя правила (rule) отдают рядом с сообщением: клиент может по нему подобрать свой текст, не разбирая человеческую строку. Сломанный JSON приходит как json_invalid с loc вида ("body", 1), где единица — позиция в тексте, а не поле: такому нарушению field не ставят. Как problem превращается в ответ 400 с типом application/problem+json — в статье про единый обработчик ошибок.

Какой язык выберется

Словарь сообщений выбирается на запрос, а не на процесс: один пользователь просит русский, другой английский. Разбор Accept-Language — несколько строк:

MESSAGES_BY_LANG = {"ru": MESSAGES_RU, "en": MESSAGES_EN}


def messages_for(request: Request) -> dict[str, str]:
    for part in request.headers.get("accept-language", "").split(","):
        lang = part.split(";")[0].strip()[:2].lower()
        if lang in MESSAGES_BY_LANG:
            return MESSAGES_BY_LANG[lang]
    return MESSAGES_RU

Своим правилам при этом тоже нужны два текста: PydanticCustomError("phone", "...") несёт один язык. Поэтому для многоязычного API в PydanticCustomError оставляют только имя, а тексты для своих типов кладут в те же словари, рядом со встроенными.

Вне запроса — в обработчике сообщения из очереди, в фоновой задаче — заголовка нет, и берут язык по умолчанию. Поэтому в машинных контурах на локализованные сообщения не опираются: там читают rule и field.

Валидатор строится один раз

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

Зависимости для своих правил передают замыканием: правило coupon, которому нужен формат из настроек, получает его при создании типа, а не читает из глобальной переменной внутри функции:

def coupon_rule(rules: CouponFormatRules):
    def check(value: str) -> str:
        if not rules.matches(value):
            raise PydanticCustomError("coupon", "неверный формат купона")
        return value
    return check


Coupon = Annotated[str, AfterValidator(coupon_rule(settings.coupons))]

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

Когда лучше проверить в сервисе

Своё правило — правильный инструмент, если правило:

  • чисто синтаксическое (формат, диапазон, структура данных);
  • переиспользуется в нескольких местах;
  • не требует обращения к базе или внешнему сервису.

Если проверка требует запроса к базе («логин уже занят», «категория существует»), её место в сценарии. У Pydantic есть и техническая причина, которой нет в других стеках.

Валидаторы синхронные. Объявить async def под @field_validator можно, но Pydantic его не ждёт: значением поля станет объект корутины, а в журнале появится RuntimeWarning: coroutine ... was never awaited. Синхронный же запрос к базе из валидатора в приложении на asyncio остановит весь цикл событий на время запроса: пока проверяется один логин, не обслуживается никто.

Гарантии всё равно нет. Между проверкой и вставкой проходит время, и за него параллельный запрос успевает занять тот же логин. Защищает только ограничение UNIQUE в базе; валидатор даёт лишь красивое сообщение.

И статус другой. «Логин занят» — не ошибка формы, а конфликт состояния: правильный ответ 409, а не 400. Все нарушения Pydantic собираются в один RequestValidationError, то есть в один ответ о неверном запросе. Значит, проверка уникальности живёт в сценарии, а её результат — Conflict, который обработчик превращает в 409. Разбор уровней — в статье про где валидировать.

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

Дополнительно: при первом чтении можно пропустить

Глубже: обязательность, None и строгостьрасширенное

Обязательность в Pydantic задаётся значением по умолчанию, а не типом. qty: int | None без умолчания — поле обязательное, но может быть null; пропустили его в запросе — получите missing. qty: int | None = None — поле необязательное. Отсюда частая ошибка: разработчик пишет | None, считая, что сделал поле необязательным, а клиент получает missing.

По умолчанию Pydantic приводит типы: строка "1" в поле int молча станет единицей. ConfigDict(strict=True) отключает приведение, и та же строка даёт int_type с текстом Input should be a valid integer; для публичного API, где "1" и 1 — разные вещи для другого клиента, строгий режим включают сознательно.

Списки проверяются поэлементно без дополнительных настроек: items: list[int] с элементом "x" даст ошибку с loc ("items", 1), а Field(min_length=1) на списке отвергнет пустой с типом too_short. Лишние поля в теле по умолчанию игнорируются; extra="forbid" превращает их в extra_forbidden, и это стоит включать на моделях запросов, чтобы опечатка в имени поля не проходила молча.

Коротко

  • Своё правило — функция плюс Annotated[str, AfterValidator(...)] как переиспользуемый тип; правило возвращает значение или бросает PydanticCustomError с именем типа, регулярное выражение компилируется один раз, пустоту отдают умолчанию None.
  • Межполевая проверка: для равенства — field_validator второго поля с info.data и проверкой наличия первого; для сложного — model_validator(mode="after"), поле для клиента — через ctx.
  • Валидатор модели выполняется только при валидных полях: лишнего сообщения не бывает, но межполевые ошибки клиент видит со второй попытки.
  • Pydantic не переводит тексты: клиенту отдают type как rule, а сообщения берут из словаря по type с подстановкой ctx; свои типы несут текст в месте объявления.
  • Имена полей в loc уже идут по псевдониму (confirmPassword), отдельной настройки не нужно; populate_by_name=True оставляют для кода и тестов.
  • Список нарушений собирается одной функцией из exc.errors(): первый элемент loc отбрасывают, json_invalid остаётся без поля; ответ 400 отдаёт единый обработчик.
  • Словарь сообщений выбирают на запрос по Accept-Language; вне запроса — язык по умолчанию, и машинные контуры читают rule, а не текст.
  • Схема модели строится при импорте, зависимости правил — замыканием; из правил не ходят в базу: валидаторы синхронные и останавливают цикл событий, гарантии нет, а «занято» — это 409.
  • Обязательность задаёт умолчание, а не тип: int | None без умолчания обязателен; strict=True отключает приведение "1" в 1; списки проверяются поэлементно, лишние поля запрещает extra="forbid".

Что почитать дальше