ORM-модели на SQLAlchemy 2.0 — DeclarativeBase, Mapped, типы

Как объявлять ORM-модели через DeclarativeBase и Mapped, правильные типы для денег, времени и UUID, почему ORM-модель ≠ агрегат ≠ Pydantic-DTO и где держать модели.

Опубликовано 23 июня 2026 ~5 мин чтения Автор: Вадим Викулин

Опирается на правила: R-SQLA-MODEL-1 … R-SQLA-MODEL-3 и R-SQLA-MODEL-X1 из SQLAlchemy Style Guide → раздел 2. ORM-модели.

Важно знать

ORM-модели живут в adapters/out/persistence/ — не в core/, не в routers/.

Три разных типа: ORM-модель (анемичная persistence-структура), доменный агрегат (core/<bc>/domain/), Pydantic-DTO (API-граница).

Деньги — Numeric(p, s) → Decimal, не Float. Время — DateTime(timezone=True) → aware datetime. UUID — UUID(as_uuid=True).

Доменная логика и инварианты на ORM-модели — запрещены. Модель только хранит строку в БД.

Маппинг модели в агрегат и обратно — явные функции to_domain / to_model рядом с репозиторием.

lazy="raise" на relationship по умолчанию: в async ленивая загрузка вызывает MissingGreenlet.

expire_on_commit=False на AsyncSession при работе с async — иначе обращение к атрибутам после commit детонирует.

ORM-модель в SQLAlchemy — это persistence-деталь: она описывает, как данные агрегата лежат в таблице PostgreSQL. Она анемична — без методов, без инвариантов, без событий. Весь смысл в том, чтобы domain-слой не знал о DeclarativeBase, а persistence-слой не знал о бизнес-правилах. Эта статья раскрывает раздел 2 SQLAlchemy Style Guide.

Размещение в гексагональной структуре

R-SQLA-MODEL-1: ORM-модели — в adapters/out/persistence/, никогда в core/.

src/
  core/
    order/
      domain/
        order.py          ← доменный агрегат Order
      port/
        order_repository.py  ← Protocol-порт
  adapters/
    out/
      persistence/
        order/
          order_model.py  ← ORM-модель (сюда)
          order_mapper.py ← to_domain / to_model
          sqla_order_repository.py

Domain-пакет не знает о SQLAlchemy, нет ни одного from sqlalchemy import ... в core/. Это проверяемо: grep -r "sqlalchemy" src/core/ должен молчать.

DeclarativeBase и Mapped

SQLAlchemy 2.0 принёс полностью аннотированный стиль через DeclarativeBase и Mapped. Все модели сервиса наследуются от одного базового класса.

# adapters/out/persistence/base.py
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

Объявление модели для агрегата Order:

# adapters/out/persistence/order/order_model.py
from datetime import datetime
from decimal import Decimal
from uuid import UUID

from sqlalchemy import DateTime, ForeignKey, Numeric, String
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship

from adapters.out.persistence.base import Base


class OrderModel(Base):
    __tablename__ = "orders"

    id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
    customer_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
    status: Mapped[str] = mapped_column(String(32), nullable=False)
    total_amount: Mapped[Decimal] = mapped_column(Numeric(19, 4), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    confirmed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    items: Mapped[list["OrderItemModel"]] = relationship(
        "OrderItemModel",
        back_populates="order",
        lazy="raise",
        cascade="all, delete-orphan",
    )

Mapped[T] — это аннотация типа, SQLAlchemy 2.0 читает её и строит Column без явного Column(...). Результат: IDE и mypy понимают типы, нет «магических» строковых аннотаций.

Правильные типы для денег, времени и UUID

R-SQLA-MODEL-2 задаёт три правила на типы — они напрямую связаны с pg-types (PG-T-013/030/040).

Деньги — Numeric, не Float

# ПРАВИЛЬНО
total_amount: Mapped[Decimal] = mapped_column(Numeric(19, 4), nullable=False)
unit_price: Mapped[Decimal] = mapped_column(Numeric(12, 4), nullable=False)

# НЕПРАВИЛЬНО — Float накапливает ошибки округления
total_amount: Mapped[float] = mapped_column(Float, nullable=False)

Numeric(19, 4) → PostgreSQL numeric(19, 4). SQLAlchemy возвращает Decimal, Python работает с точной арифметикой. Колонка с Float может дать 999.9999999999 вместо 1000.00 на уровне агрегата — критично для любого расчёта суммы.

Время — DateTime(timezone=True), не naive datetime

# ПРАВИЛЬНО — aware datetime (UTC)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

# НЕПРАВИЛЬНО — naive datetime теряет timezone
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)

DateTime(timezone=True) → PostgreSQL timestamptz. SQLAlchemy возвращает aware datetime (UTC). Naive datetime теряет информацию о зоне и вызывает трудноотлаживаемые расхождения при сравнении дат между сервисами.

UUID — UUID(as_uuid=True)

from sqlalchemy.dialects.postgresql import UUID as PG_UUID

id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)

as_uuid=True → SQLAlchemy возвращает Python uuid.UUID, а не строку. Без этого флага маппер получает строку и обязан парсить её вручную на каждой операции чтения. PostgreSQL хранит UUID эффективно (16 байт), а не как varchar(36).

Три разных типа: ORM-модель ≠ агрегат ≠ Pydantic-DTO

R-SQLA-MODEL-3 — самое важное концептуальное правило.

OrderModel       ← SQLAlchemy ORM, persistence-деталь в adapters/out/persistence/
Order            ← доменный агрегат в core/order/domain/, с методами confirm(), cancel()
OrderResponse    ← Pydantic BaseModel в routers/, сериализуется в JSON API

Сводить их в один тип — соблазнительная «экономия», которая ломает всё:

ORM-модель как агрегат: handler получает объект с InstrumentedAttribute, без доменных методов; SQLAlchemy-сессия начинает трекать изменения там, где не ожидалось.
Pydantic-модель как ORM: model_validate из dict работает для API, но не описывает отношения, lazy loading, identity map.
Агрегат как Pydantic: нельзя иметь @property с инвариантом и model_validate одновременно без хаков.

Каждый тип несёт одну ответственность. Маппинг между ними — явный, локализованный.

Relationship и eager loading

В async SQLAlchemy ленивая загрузка (lazy="select") не работает: попытка обратиться к lazy-атрибуту вне активной сессии выдаёт MissingGreenlet. Поэтому — lazy="raise" по умолчанию, а нужные связи загружаются явно в запросе.

# order_item_model.py
class OrderItemModel(Base):
    __tablename__ = "order_items"

    id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
    order_id: Mapped[UUID] = mapped_column(
        PG_UUID(as_uuid=True),
        ForeignKey("orders.id", ondelete="CASCADE"),
        nullable=False,
    )
    product_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
    quantity: Mapped[int] = mapped_column(nullable=False)
    unit_price: Mapped[Decimal] = mapped_column(Numeric(12, 4), nullable=False)

    order: Mapped["OrderModel"] = relationship(
        "OrderModel",
        back_populates="items",
        lazy="raise",
    )

В репозитории при загрузке агрегата с items:

from sqlalchemy import select
from sqlalchemy.orm import selectinload

stmt = (
    select(OrderModel)
    .options(selectinload(OrderModel.items))
    .where(OrderModel.id == order_id)
)
result = await session.execute(stmt)
model = result.scalar_one_or_none()

selectinload выдаёт два запроса: один за orders, один за всеми order_items для найденных заказов. joinedload даёт один запрос с JOIN, но умножает строки при многих items. Выбор зависит от объёма — для небольших коллекций selectinload предпочтительнее.

Конфигурация AsyncSession

# adapters/out/persistence/session.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(settings.database_url, echo=False, pool_size=10)

SessionFactory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
    autocommit=False,
)

expire_on_commit=False критично в async: при дефолтном expire_on_commit=True SQLAlchemy помечает все атрибуты объекта «устаревшими» после commit(). При следующем обращении ORM пытается сделать SELECT, но async-сессия уже закрыта → MissingGreenlet. Решение: маппи ORM-объект в доменный агрегат до выхода из транзакции (что и делает маппер to_domain).

Модель Customer с JSONB-атрибутами

# adapters/out/persistence/customer/customer_model.py
from sqlalchemy import String, Text
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID

class CustomerModel(Base):
    __tablename__ = "customers"

    id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
    email: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
    full_name: Mapped[str] = mapped_column(String(512), nullable=False)
    address: Mapped[dict] = mapped_column(JSONB, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

JSONB — PostgreSQL-специфичный тип, хранится бинарно, поддерживает GIN-индексы. Если поле всегда nullable — Mapped[dict | None].

Что запрещено

Антипаттерн	Правило	Что взамен
ORM-модель в `core/` или в `routers/`	`R-SQLA-MODEL-1`	Только в `adapters/out/persistence/<bc>/`
`Float` для денег	`R-SQLA-MODEL-2`	`Numeric(p, s)` → `Decimal`
`DateTime` без `timezone=True` для дат с зоной	`R-SQLA-MODEL-2`	`DateTime(timezone=True)` → aware `datetime`
`UUID(as_uuid=False)` → строка в Python	`R-SQLA-MODEL-2`	`UUID(as_uuid=True)` → `uuid.UUID`
Доменные методы / инварианты на ORM-модели	`R-SQLA-MODEL-X1`	Методы и инварианты — на агрегате в `core/`
ORM-модель как Pydantic-схема или как агрегат	`R-SQLA-MODEL-3`	Три отдельных типа + явный маппер
`lazy="select"` на relationship в async	`R-SQLA-QRY-X2`	`lazy="raise"` + `selectinload`/`joinedload` в запросе

Куда дальше

python/repository-pattern.md — как Protocol-порт в core/ и SqlAlchemy<X>Repository в adapters/out/persistence/ используют эти модели; правила R-SQLA-REPO-*.
python/mapping.md — явные функции to_domain / to_model, сборка агрегата из нескольких моделей, почему __dict__ и vars() запрещены; правила R-SQLA-MAP-*.
python/session-transactions.md — граница транзакции на Handler через Unit of Work, expire_on_commit=False в контексте UoW, обработка ошибок; правила R-SQLA-SESS-*.