SQLModel и Pydantic: как описать ограничения домена на уровне схем, а не после запроса

Практически в любой прикладной системе со временем появляется знакомая проблема: «мы проверяем данные уже после того, как они оказались в базе». Часто это выглядит так: приложение делает INSERT, база принимает строку/число без возражений, а затем код пытается поймать бизнес-ошибки вручную — то ли при чтении, то ли перед обработкой. В результате:

• в таблицах накапливаются неконсистентные записи;
• логика ограничений размазывается по сервисам и эндпоинтам;
• ошибки становятся несогласованными (где-то 400, где-то 500, где-то вообще молча “проглатывается”);
• поддержка превращается в расследование: «а почему в этой записи поле X не соответствует домену?»

Современный стек на Python даёт достаточно инструментов, чтобы приблизить валидацию к модели и схеме данных. В этой статье разберём подход, в котором ограничения домена живут ближе к модели: через типы, поля, предикаты и согласованные ошибки — до того, как запрос ушёл в БД или сразу при попытке её обновить.

Фокус — на связке SQLModel и Pydantic: как использовать их, чтобы формально описывать доменные инварианты на уровне схемы (и получать предсказуемое поведение), а не «лечить» данные после запроса.

Почему “валидация после запроса” — системная проблема

Представьте типичную архитектуру:

1. API принимает JSON.
2. Сериализатор преобразует в ORM-модель или в Pydantic-объект.
3. Код делает запрос в БД.
4. Если данные неверные, приложение начинает “отрабатывать” исключения или возвращать сообщения уже после факта.

Это уже лучше, чем “валидируем только на клиенте”, но всё равно недостаточно. Причина не только в качестве данных. Валидация после запроса означает, что:

• нет строгой точки истины для доменных правил;
• часть правил может быть реализована на стороне приложения, часть — на стороне базы, часть — нигде;
• появляются гонки и крайние случаи (например, два потока одновременно пишут разные значения в поле, а проверка ограничений выполняется позже);
• при миграциях и изменениях домена вы начинаете искать места, где правила дублировались.

Более устойчивый подход — строить схему так, чтобы несоответствие домену обнаруживалось как можно раньше, желательно:

• на этапе создания/обновления модели в Python;
• и/или на уровне БД через ограничения (check/unique/foreign key);
• и/или на уровне SQLModel/Pydantic при сборке данных.

SQLModel здесь интересен тем, что объединяет два мира: декларативную модель БД (как у ORM) и декларативную схему данных (как у Pydantic). Поэтому большую часть “доменных” проверок можно описать прямо в модели: типы, диапазоны, паттерны, предикаты и валидаторы.

Как устроен SQLModel поверх Pydantic: что можно “переместить” ближе к модели

SQLModel использует Pydantic как основу для валидации. Это означает:

• типы и поля модели — это не просто аннотации для IDE, это материал для валидации входных данных;
• механизмы Pydantic (например, Field, валидаторы, Annotated, кастомные типы) работают и в SQLModel;
• ошибки валидации можно получать в стандартном формате Pydantic, что упрощает консистентность API.

Однако есть важный нюанс: часть ограничений реально может быть выражена на уровне SQL (например, CHECK, UNIQUE), а часть — только на уровне Python. Идея статьи — максимально приблизить ограничения к модели, но понимать границу: если правило должно быть непротиворечивым между всеми источниками записи (несколько сервисов, миграционные скрипты, прямые SQL-запросы), то его лучше закреплять и в БД. А если это правило специфично для поведения приложения и связано с преобразованиями/инвариантами на стороне данных, то Pydantic-валидация в модели будет особенно полезна.

С практической точки зрения строится “слоистая” защита:

1. Pydantic/SQLModel валидирует структуру и часть инвариантов ещё до INSERT/UPDATE.
2. БД ограничивает критичные инварианты на уровне данных.
3. Приложение обрабатывает ошибки предсказуемо и единообразно.

Модель домена: базовый пример полей и диапазонов

Рассмотрим домен: пользователь создаёт заказ. В нём есть поля total_amount (неотрицательная сумма) и currency (только из допустимого набора). В традиционном стиле ограничения часто проверяются после запроса или отдельным кодом.

С SQLModel/Pydantic правила можно выразить прямо в модели:

from typing import Optional, Literal
from sqlmodel import SQLModel, Field
from pydantic import validator

class OrderBase(SQLModel):
    total_amount: float = Field(..., ge=0)
    currency: Literal["RUB", "USD", "EUR"] = "RUB"

class Order(OrderBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

Что мы получили:

• total_amount с ge=0 — валидация диапазона на стороне Pydantic.
• currency — статический набор значений (Literal), который Pydantic проверяет при парсинге входных данных.

Теперь важно: если приложение пробует создать заказ с отрицательной суммой, ошибка возникает до записи в БД — то есть “невалидные” записи не попадают внутрь таблицы.

Типичные ошибки здесь

1. Использовать float для денег без мысли о точности.
   float даст проблемы округления. Правильнее использовать decimal.Decimal и валидировать через condecimal или через custom type. Если вам действительно важна финансовая точность — лучше сразу перейти на Decimal.

2. Оставлять “доменные” правила только в приложении.
   Даже если Pydantic защищает API, внешние вставки (скрипты миграций, прямые SQL запросы) могут обойти правила. Для критичных инвариантов — добавляйте ограничения и в БД (ниже вернёмся).

Инварианты “между полями”: валидаторы и предикаты

Часто правило выглядит не как ограничение одного поля, а как согласованность нескольких: например, start_date должна быть раньше end_date, или min_age <= max_age.

Pydantic даёт валидаторы, которые можно встроить в SQLModel.

Пример: период активности:

from datetime import date
from typing import Optional
from sqlmodel import SQLModel, Field
from pydantic import model_validator

class Period(SQLModel):
    start_date: date
    end_date: date

    @model_validator(mode="after")
    def check_order(self):
        if self.end_date < self.start_date:
            raise ValueError("end_date must be >= start_date")
        return self

class Subscription(Period, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

model_validator(mode="after") позволяет сравнить значения уже после того, как поля прошли базовую валидацию типов/диапазонов.

Почему это лучше, чем “проверить после запроса”

• Ошибка возникает в одном месте и имеет предсказуемый тип (валидация модели).
• Вся бизнес-логика консистентна: нет ситуации, когда одно endpoint проверяет, а другое — нет.
• При тестировании вы можете проверять правило как функцию построения модели, а не как побочный эффект от SQL.

Подводный камень: где именно вы ловите ошибки

Если вы создаёте SQLModel-объект внутри endpoint’а, то Pydantic-ошибки обычно всплывают как ValidationError. Но если вы создаёте объект, затем делаете session.add(), а валидация “ждёт” до commit, поведение может отличаться в зависимости от конфигурации и того, где именно вы создаёте объект.

Рекомендация: создавайте и валидируйте модель в Python до того, как выполняете SQL. То есть: Order.model_validate(...) или прямое создание Order(...) на основе входных данных — до формирования SQL-сессии.

Согласованные ошибки: что делать с доменными исключениями

Проблема “после запроса” часто выражается в хаотичных ошибках: иногда вы получаете SQLAlchemy exception, иногда — Pydantic, иногда — кастомные сообщения.

С SQLModel/Pydantic можно добиться консистентности через:

1. Использование ValueError/TypeError в валидаторах.
2. Преобразование ошибок в формат, который ожидает API (например, единый ответ {field, message}).
3. Желательно — различать:
   • ошибку “данные не соответствуют домену” (обычно 400);
   • ошибку “ошибка инфраструктуры” (500/503).

Пример: превращаем доменную ошибку в понятный формат

Упрощённо для иллюстрации:

from fastapi import HTTPException
from pydantic import ValidationError

def translate_validation_error(err: ValidationError):
    # err.errors() -> список структур с loc/type/msg
    # Соберём в компактный формат
    details = []
    for e in err.errors():
        loc = ".".join(str(x) for x in e.get("loc", []))
        details.append({"field": loc, "message": e.get("msg")})
    return details

def validate_or_400(model_cls, data):
    try:
        return model_cls.model_validate(data)
    except ValidationError as e:
        details = translate_validation_error(e)
        raise HTTPException(status_code=400, detail=details)

Дальше в endpoint’е вы можете делать:

order = validate_or_400(Order, payload)
session.add(order)
session.commit()

Так вы гарантируете: доменные ошибки будут стабильно “приклеены” к валидации модели.

Ограничения на уровне схемы БД: CHECK и уникальность как дополнительный слой

Pydantic может не знать, как именно должна выглядеть “истина” на уровне БД. Поэтому реальная устойчивость достигается комбинацией: валидация модели + ограничения БД.

SQLModel позволяет объявлять некоторые ограничения через Field (и/или через возможности SQLAlchemy под капотом). Для примера рассмотрим уникальность по паре полей: например, пользователь не может иметь два активных периода с одинаковым start_date.

from datetime import date
from typing import Optional
from sqlmodel import SQLModel, Field
from sqlalchemy import UniqueConstraint

class PeriodSubscription(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    start_date: date
    end_date: date

    __table_args__ = (
        UniqueConstraint("user_id", "start_date", name="uq_user_start"),
    )

Теперь даже при гонках (два запроса одновременно) БД гарантирует целостность. А SQLModel/Pydantic может предотвратить часть нарушений до SQL.

Где CHECK действительно полезен

Для “числовых инвариантов” CHECK на уровне БД — лучший вариант, если важно, чтобы некорректные данные не могли появиться ни откуда.

Например, total_amount не должен быть отрицательным. Если вы уже сделали ge=0 в Pydantic — вы неплохо защищены на уровне API, но CHECK в БД добавляет гарантию.

В SQLAlchemy/SQLModel обычно это делается через аргументы колонки (зависит от версии и способа объявления). Концептуально идея такая: у вас должно быть правило, которое БД enforce’ит всегда.

Встроенные предикаты для полей: когда достаточно Field()

С большинством “атомарных” ограничений можно обойтись Field(...) и встроенными параметрами:

• ge, le для чисел,
• min_length, max_length для строк,
• pattern для регулярных выражений (с осторожностью),
• default_factory и прочее.

Пример: SKU или идентификатор заказа должен соответствовать формату.

from sqlmodel import SQLModel, Field
from typing import Optional

class Product(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    sku: str = Field(
        ..., 
        min_length=3, 
        max_length=30,
        pattern=r"^[A-Z0-9\-]+$"
    )

Плюсы:

• правила декларативные и читаемые;
• ошибки Pydantic структурированы;
• вы снижаете вероятность “проскочивших” некорректных значений.

Минусы и подводные камни:

• pattern — это regex. Регулярки в валидации иногда становятся источником “расхождений” между тем, что вы ожидаете, и тем, что реально передаёт фронт.
• Для сложных правил (например, “SKU уникален в рамках магазина и должен проходить алгоритм контрольной суммы”) лучше выносить в валидатор.

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

Если у вас есть доменные типы, которые встречаются в разных моделях: например, Email, PhoneNumber, Money, NonEmptyString, то простая Field() валидация может оказаться недостаточной.

Кастомный тип (или конструкция на базе Pydantic) позволяет:

• централизовать правила;
• обеспечить единый формат хранения/парсинга;
• использовать везде один и тот же механизм валидации.

Пример для Money с Decimal (упрощённо):

from decimal import Decimal, InvalidOperation
from typing import Any
from pydantic_core import core_schema
from pydantic import GetCoreSchemaHandler
from pydantic import BaseModel

class Money(Decimal):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler):
        # Упрощение: в реальном проекте можно построить schema более тонко,
        # но суть — в централизованной конвертации/валидации.
        return core_schema.no_info_plain_validator_function(cls.validate)

    @classmethod
    def validate(cls, v):
        try:
            d = Decimal(str(v))
        except (InvalidOperation, TypeError):
            raise ValueError("Invalid money amount")
        if d < 0:
            raise ValueError("Money must be >= 0")
        return d

Дальше:

from sqlmodel import SQLModel, Field
from typing import Optional

class Invoice(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    total: Money = Field(...)

Смысл: доменное правило “деньги >= 0” закреплено в типе, а не продублировано в каждом endpoint’е.

На практике корректную схему для кастомных типов удобнее делать через механизмы Pydantic v2 (как выше) или через pydantic-провайдеры. Важно, чтобы тип был единым источником истины.

Обновления и частичные PATCH: как не потерять инварианты

Особенно часто “валидация после запроса” появляется из-за PATCH: когда вы обновляете только одно поле, а валидатор меж-полевых инвариантов требует оба значения.

Решение: разделяйте DTO для создания и DTO для обновления, и проектируйте валидаторы с учётом режима частичного обновления.

Пример: при PATCH пользователь может прислать start_date или end_date не сразу. Если вы валидируете модель без недостающих полей, сравнение “end >= start” будет невозможно.

Подходы:

1. Использовать отдельные модели:
   • SubscriptionCreate — все поля обязательны.
   • SubscriptionUpdate — поля опциональны.
   • меж-полевую проверку делать на уровне сервиса, когда у вас есть текущие значения из БД.
2. Валидировать “на основе полного состояния”: собирать итоговую модель (текущая + патч), затем запускать model_validator.

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

• загрузить текущую запись;
• применить patch в объект/словарь;
• построить “итоговую” модель домена (где валидаторы меж-полей сработают);
• только потом выполнять UPDATE.

Так вы сохраняете принцип “инварианты живут ближе к модели”, но честно учитываете, что на вход в PATCH может быть не полный набор данных.

Мини-кейс: ограничения домена “до INSERT”, а не после commit

Возьмём простую сущность: “пользователь может иметь только один активный профиль по ключу email”. Допустим, “активный” означает is_active=True.

Нюанс: это уже похоже на правило уровня данных/уникальности с условием. В реляционных БД это часто решается partial unique index, но не везде есть поддержка.

В SQLModel/Pydantic мы можем хотя бы:

• валидировать формат email;
• проверять согласованность статуса при создании;
• и потом падать корректной ошибкой при конфликте уникальности.

Структура может быть такая:

from sqlmodel import SQLModel, Field
from typing import Optional
from pydantic import EmailStr
from datetime import datetime

class UserProfileBase(SQLModel):
    email: EmailStr
    is_active: bool = True

class UserProfile(UserProfileBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)

А сервисная логика (согласованность “единственного активного”) обычно требует запроса к