Валидация доменных правил в Python: от Pydantic схем до согласованных ошибок для API
Разберём, как отделить формальную валидацию типов от бизнес-правил, где держать логику ограничений и как возвращать ошибки клиенту так, чтобы их можно было обработать одинаково во всех эндпоинтах.
Содержание
Валидация доменных правил в Python: от Pydantic-схем до согласованных ошибок для API
Валидация входящих данных — та часть API, которая чаще всего «растёт вместе с продуктом» и постепенно превращается в источник хаоса. На раннем этапе хватает проверки типов и пары if-ов, но когда появляются бизнес-ограничения (например, «статус нельзя сменить на архивный из любого места», «код товара обязан быть уникален в конкретном складе», «дата окончания не может быть раньше начала»), логика начинает размазываться по эндпоинтам. Результат — разные форматы ошибок, неодинаковая семантика сообщений, сложнее поддерживать фронтенд и интеграции.
Задача этой статьи — показать практичный подход: как отделить формальную валидацию типов от доменных правил, как организовать ограничения в коде, и как возвращать ошибки клиенту согласованно во всех эндпоинтах. В качестве инструмента возьмём Pydantic (в экосистеме FastAPI) и рассмотрим, как выстроить «контракт» ошибок, удобный и для клиентов, и для разработчиков.
Разделение ответственности: типы, домен и транспорт
Чтобы не утонуть в условностях, полезно начать с простого разбиения обязанностей.
Валидация типов и структуры (формальная)
Это проверки «на форму» входных данных:
- входной JSON должен соответствовать ожидаемой структуре;
- типы полей корректны (строка/число/дата);
- обязательные поля присутствуют;
- базовые форматы (например, email/UUID) корректны.
Для этого обычно достаточно Pydantic-схем (моделей запроса/ответа), особенно если вы используете FastAPI.
Доменные правила (содержательная)
Это проверки «на смысл»:
- «количество товара не может быть отрицательным» (может быть доменное, даже если
int— валиден как тип); - «статус заказа нельзя перевести в X, если текущий Y»;
- «цена должна быть больше нуля, если категория требует цены»;
- «у пользователя должен быть доступ к ресурсу, указанному в запросе».
Ключевой момент: доменные правила зависят от контекста и состояния (например, от текущего состояния заказа). Типовой вопрос: «где держать эту логику?». Если всё свалить в Pydantic, вы получите неудобства:
- трудно тестировать доменные правила отдельно от HTTP;
- появляется зависимость схем от окружения (БД, сервисов);
- растёт магия в
@validator, а правила становятся трудно читаемыми.
Оптимальная архитектура: схемы отвечают за структуру и базовые ограничения, а доменные правила — за смысл и консистентность. В идеале доменная логика не должна знать о FastAPI и HTTP.
Транспорт и представление ошибок (API contract)
API должен возвращать ошибки клиенту так, чтобы:
- фронтенд и интеграции могли одинаково распознавать ошибки;
- ошибки были предсказуемыми по форме (stable contract);
- было видно, какое поле нарушено, и какое правило не выполнено;
- ошибки разного происхождения (Pydantic, бизнес, авторизация) попадали в единый формат.
Pydantic-схемы: где заканчивается «структура» и начинается домен
Начнём с базовой схемы запроса. Допустим, у нас сервис создания заказа. Есть поля:
customer_id— UUID клиента;items— список позиций;currency— валюта;note— опциональное примечание.
from uuid import UUID
from decimal import Decimal
from typing import List, Optional, Literal
from pydantic import BaseModel, Field, conlist, ConfigDict
class OrderItemIn(BaseModel):
product_id: UUID
quantity: int = Field(gt=0, description="Количество должно быть положительным")
unit_price: Decimal = Field(gt=0, description="Цена за единицу должна быть больше нуля")
class OrderCreateIn(BaseModel):
customer_id: UUID
items: conlist(OrderItemIn, min_length=1) # хотя это похоже на бизнес, обычно это структурное ограничение
currency: Literal["RUB", "USD", "EUR"] = "RUB"
note: Optional[str] = Field(default=None, max_length=500)
model_config = ConfigDict(extra="forbid")
Здесь quantity > 0, unit_price > 0, note ограничен по длине — это, как правило, «формальные» ограничения, которые не требуют обращения к БД и не зависят от контекста. Их можно держать в Pydantic.
Валидация на уровне схемы: single responsibility
Pydantic удобен тем, что умеет собирать ошибки по полям. Но важно помнить: @model_validator и @field_validator в Pydantic легко превращаются в «мини-движок домена», если туда перенести сложные правила.
Хороший критерий:
- Если правило не требует внешних данных и является чистой функцией от входа — держите в схеме.
- Если правило зависит от текущего состояния, доступов, справочников, БД — выносите из схемы.
Межполевая валидация: допустимо, но осторожно
Например, правило: если валюта указана USD, то сумма заказа должна быть не меньше минимального порога. Но порог зависит от курса и конфигурации — значит, это не чистая валидация. А если порог статический и известен? Тогда можно.
Для чистых межполей в Pydantic полезен @model_validator:
from pydantic import model_validator
class OrderCreateIn(BaseModel):
# ... поля как выше ...
@model_validator(mode="after")
def check_note_required_for_large_orders(self) -> "OrderCreateIn":
total = sum(i.quantity * i.unit_price for i in self.items)
if total > Decimal("10000") and not self.note:
raise ValueError("note is required for orders above 10000")
return self
Но это правило уже похоже на бизнес (хотя не требует внешних данных). Его можно оставить в схеме, если вы сознательно принимаете такую связку. Однако практика показывает: чем больше доменных правил — тем лучше они живут в отдельном сервисном слое.
Доменные правила: как выстроить слой, который живёт без HTTP
Теперь перейдём к домену. Предположим, есть доменная сущность Order и слой приложения (use-case), который создаёт заказ.
Вариант 1: доменная валидация как исключения с кодами
Можно сделать доменные проверки отдельными функциями/классами, которые возбуждают предсказуемые исключения. Например:
- ошибки бывают разных типов (
DomainRuleViolation); - у ошибки есть
code(стабильный идентификатор правила),message(человекочитаемое),field(опционально).
from dataclasses import dataclass
from typing import Optional, List
@dataclass
class FieldViolation:
field: Optional[str]
code: str
message: str
class DomainValidationError(Exception):
def __init__(self, violations: List[FieldViolation]):
self.violations = violations
super().__init__("Domain validation error")
Пример доменных правил: допустим, в компании запрещены заказы, если сумма позиции по одному продукту превышает лимит, зависящий от категории продукта — а категория определяется из справочника. Тогда доменный слой может принимать контекст (например, репозиторий) и выполнять проверки.
from decimal import Decimal
from uuid import UUID
class ProductRepository:
def get_category(self, product_id: UUID) -> str:
# заглушка
return "standard"
class LimitsPolicy:
def max_total_by_category(self, category: str) -> Decimal:
return Decimal("50000") if category == "standard" else Decimal("20000")
def validate_order_domain(
customer_id: UUID,
items: list,
repo: ProductRepository,
policy: LimitsPolicy,
) -> None:
violations: List[FieldViolation] = []
total = sum(i.quantity * i.unit_price for i in items)
if total <= 0:
violations.append(FieldViolation(
field=None,
code="ORDER_TOTAL_NON_POSITIVE",
message="Total order amount must be positive"
))
for idx, item in enumerate(items):
category = repo.get_category(item.product_id)
max_total = policy.max_total_by_category(category)
item_total = item.quantity * item.unit_price
if item_total > max_total:
violations.append(FieldViolation(
field=f"items[{idx}].unit_price",
code="ITEM_TOTAL_EXCEEDS_LIMIT",
message=f"Item total exceeds limit for category={category}"
))
if violations:
raise DomainValidationError(violations)
Обратите внимание: доменная функция не знает о JSON, не знает о FastAPI. Она возвращает ошибки в удобной форме для последующего трансформационного слоя (см. ниже).
Вариант 2: «результат» вместо исключений
Для некоторых команд исключения кажутся «дорогими» или слишком «магическими». Тогда можно использовать Result-подход: возвращать структуру с ok/data или errors. Но на практике в Python исключения остаются наиболее привычным механизмом для валидации. Главное — стандартизировать обработку.
Согласованный формат ошибок для API
Теперь важнейшая часть: как привести к одному виду ошибки Pydantic и доменные ошибки, чтобы клиенту не приходилось писать отдельный парсер под каждый источник проблем.
Контракт ошибки (рекомендуемая структура)
Предложим стандарт для ошибок:
{
"error": {
"type": "validation_error",
"message": "Request validation failed",
"details": [
{
"field": "items[0].quantity",
"code": "VALUE_GT",
"message": "quantity must be greater than 0"
}
]
}
}
type— общий класс ошибки (validation_error,unauthorized,forbidden,conflict).message— кратко для логов и UI.details— список нарушений; каждый пункт содержитfield(илиnull),codeиmessage.
Клиенту удобно обрабатывать:
- наличие
type; - список
details; - соответствие
codeвалидационным сценариям.
Приведение ошибок Pydantic
В FastAPI ошибки Pydantic обычно представлены как RequestValidationError, с полем errors(), где есть loc, msg, type. Мы можем преобразовать это в наш контракт.
Проблема: loc по умолчанию может быть в формате ("body","items",0,"quantity"). Нужно научиться маппить на читабельные field.
Вот пример маппинга:
from typing import Any, Dict, List, Tuple, Optional
def loc_to_field(loc: Tuple[Any, ...]) -> Optional[str]:
# Простейшая стратегия: отбрасываем "body" и собираем остальное
parts = []
for x in loc:
if x == "body":
continue
if isinstance(x, int):
# индекс списка
if parts:
parts[-1] = f"{parts[-1]}[{x}]"
else:
parts.append(f"[{x}]")
else:
parts.append(str(x))
# Убираем пустые
field = ".".join(p for p in parts if p and p != "0")
return field or None
Тогда преобразование:
def pydantic_errors_to_details(errors: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
details = []
for e in errors:
loc = tuple(e.get("loc", ()))
details.append({
"field": loc_to_field(loc),
"code": str(e.get("type")), # например, "value_error.number.not_gt"
"message": e.get("msg"),
})
return details
Согласование с доменными ошибками
Доменные ошибки мы уже накопили как violations с code/message/field. Теперь нужно вернуть их в том же формате.
Пример: FastAPI роутер + единый обработчик ошибок
Ниже — цельная схема, показывающая, как собрать Pydantic-валидацию и доменные правила в один ответ.
1) Модели и доменные исключения
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from uuid import UUID
from decimal import Decimal
from typing import List, Optional, Dict, Any
class OrderItemIn(BaseModel):
product_id: UUID
quantity: int = Field(gt=0)
unit_price: Decimal = Field(gt=0)
class OrderCreateIn(BaseModel):
customer_id: UUID
items: List[OrderItemIn]
currency: str = "RUB"
note: Optional[str] = Field(default=None, max_length=500)
class DomainValidationError(Exception):
def __init__(self, violations):
self.violations = violations
super().__init__("Domain validation error")
class FieldViolation:
def __init__(self, field: Optional[str], code: str, message: str):
self.field = field
self.code = code
self.message = message
2) Use-case: доменная валидация после Pydantic
class ProductRepository:
def get_category(self, product_id: UUID) -> str:
return "standard"
class LimitsPolicy:
def max_total_by_category(self, category: str) -> Decimal:
return Decimal("50000") if category == "standard" else Decimal("20000")
def validate_order_domain(items: list, repo: ProductRepository, policy: LimitsPolicy) -> None:
violations = []
for idx, item in enumerate(items):
category = repo.get_category(item.product_id)
max_total = policy.max_total_by_category(category)
item_total = item.quantity * item.unit_price
if item_total > max_total:
violations.append(FieldViolation(
field=f"items[{idx}].unit_price",
code="ITEM_TOTAL_EXCEEDS_LIMIT",
message=f"Item total exceeds limit for category={category}"
))
if violations:
raise DomainValidationError(violations)
3) Единый формат ошибок и обработчики
from fastapi.exceptions import RequestValidationError
from starlette.status import HTTP_400_BAD_REQUEST
def error_payload(type_: str, message: str, details: List[Dict[str, Any]]):
return {
"error": {
"type": type_,
"message": message,
"details": details,
}
}
def loc_to_field(loc):
parts = []
for x in loc:
if x == "body":
continue
if isinstance(x, int):
if parts:
parts[-1] = f"{parts[-1]}[{x}]"
else:
parts.append(f"[{x}]")
else:
parts.append(str(x))
return ".".join(parts) if parts else None
def pydantic_to_details(exc: RequestValidationError) -> List[Dict[str, Any]]:
details = []
for e in exc.errors():
loc = tuple(e.get("loc", ()))
details.append({
"field": loc_to_field(loc),
"code": e.get("type"),
"message": e.get("msg"),
})
return details
Регистрируем обработчики:
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
details = pydantic_to_details(exc)
return JSONResponse(
status_code=HTTP_400_BAD_REQUEST,
content=error_payload("validation_error", "Request validation failed", details),
)
@app.exception_handler(DomainValidationError)
async def domain_validation_exception_handler(request: Request, exc: DomainValidationError):
details = []
for v in exc.violations:
details.append({
"field": v.field,
"code": v.code,
"message": v.message,
})
return JSONResponse(
status_code=HTTP_400_BAD_REQUEST,
content=error_payload("validation_error", "Domain rules validation failed", details),
)
Теперь ошибки Pydantic и доменные будут выглядеть одинаково по структуре error.details[], хотя code и message могут отличаться.
4) Эндпоинт: простой контроль потока
@app.post("/orders")
async def create_order(payload: OrderCreateIn):
repo = ProductRepository()
policy = LimitsPolicy()
# Pydantic уже гарантирует структуру и базовые типы/ограничения.
# Дальше — доменная валидация.
validate_order_domain(payload.items, repo, policy)
# Здесь — создание заказа (не показано).
return {
"id": "generated-order-id",
"customer_id": str(payload.customer_id),
"currency": payload.currency,
}
Типичные ошибки и подводные камни
1) Пытаться «всё валидировать» на уровне Pydantic
Это приводит к смешению слоёв: схема начинает зависеть от внешних данных. Признаки проблемы:
- валидаторы требуют репозитории или клиентские сервисы;
- правила становятся длинными, тестирование превращается в тестирование HTTP-слоя;
- ошибки разных типов получают разный формат (потому что вы «раскидываете» ошибки разными способами).
Решение: оставляйте Pydantic для структуры, домен — для смысловых ограничений.
2) Несогласованные коды ошибок
Если у вас часть ошибок имеет code, а часть — только message, клиенту приходится анализировать тексты (а это плохая практика). Стремитесь к тому, чтобы доменные правила имели стабильный code, не привязанный к формулировке.
Для Pydantic type обычно тоже стабилен, но семантика может быть «технической». Часто полезно переименовать code в более продуктовые
Комментарии
Пока нет комментариев