Опирается на правила: R-SQLA-MAP-1, R-SQLA-MAP-2, R-SQLA-MAP-X1, R-SQLA-MODEL-2, R-SQLA-MODEL-3 из SQLAlchemy Style Guide → раздел 3. Маппинг ORM ↔ domain.

Важно знать

  • ORM-модель (DeclarativeBase + Mapped) — анемичная persistence-структура. Она живёт в adapters/out/persistence/, не в core/.
  • Маппер — явные функции to_domain(model) -> Aggregate и to_model(aggregate) -> Model в том же файле, что и репозиторий, или в отдельном <entity>_mapper.py рядом.
  • Наружу из репозитория уходит только доменный объект. ORM-модель — внутренняя деталь.
  • Сборка агрегата с вложенными коллекциями (order.items) — целиком в маппере, не размазана по методам репозитория.
  • expire_on_commit=False на AsyncSession обязателен; маппинг в domain должен происходить до выхода из транзакции, чтобы не нарваться на MissingGreenlet при доступе к lazy-атрибутам после commit.
  • Decimal для денег, aware datetime для времени, uuid.UUID через UUID(as_uuid=True) — типы ORM-модели задают правило R-SQLA-MODEL-2.
  • __dict__/vars() и прямое использование ORM-модели как domain-объекта запрещены (правило R-SQLA-MAP-X1).

Маппер — это граница, на которой persistence-детали перестают существовать для бизнес-логики. ORM-модель отражает схему PostgreSQL: колонки, типы, foreign keys. Доменный объект отражает бизнес-инварианты: Order с confirm(), Product с apply_discount(), Customer с identity. Два этих представления живут в разных слоях и встречаются ровно в одном месте — в маппере.

Структура файлов

Маппер живёт рядом с репозиторием в adapters/out/persistence/:

adapters/out/persistence/
    models.py                   # DeclarativeBase + ORM-модели
    order_mapper.py             # to_domain + to_model
    order_repository.py         # SqlAlchemyOrderRepository
    product_mapper.py
    product_repository.py

Альтернатива — маппер внутри модуля репозитория (непубличные функции в том же файле). Выбор зависит от объёма: если маппер превышает 40 строк, выносить в отдельный <entity>_mapper.py.

ORM-модели

ORM-модели — в adapters/out/persistence/models.py, не в core/. Они знают о таблицах, типах PostgreSQL, индексах. Domain-модуль (core/) их не видит.

# adapters/out/persistence/models.py
from __future__ import annotations

import uuid
from datetime import datetime
from decimal import Decimal

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


class Base(DeclarativeBase):
    pass


class OrderModel(Base):
    __tablename__ = "orders"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    customer_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
    status: Mapped[str] = mapped_column(String(32), nullable=False)
    total_amount: Mapped[Decimal] = mapped_column(Numeric(19, 2), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

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


class OrderItemModel(Base):
    __tablename__ = "order_items"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    order_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("orders.id"), nullable=False)
    product_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
    quantity: Mapped[int] = mapped_column(nullable=False)
    unit_price: Mapped[Decimal] = mapped_column(Numeric(19, 2), nullable=False)

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

lazy="raise" на relationship гарантирует: если репозиторий забыл сделать selectinload, атрибут бросит ошибку немедленно — не молча уйдёт в N+1 и не упадёт с MissingGreenlet.

Простой случай: one-to-one маппинг

Доменные объекты для core/order/:

# core/order/entities.py
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum


class OrderStatus(str, Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"


@dataclass
class OrderItem:
    id: uuid.UUID
    product_id: uuid.UUID
    quantity: int
    unit_price: Decimal


@dataclass
class Order:
    id: uuid.UUID
    customer_id: uuid.UUID
    status: OrderStatus
    total_amount: Decimal
    created_at: datetime
    items: list[OrderItem] = field(default_factory=list)

    def confirm(self) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot confirm order in status {self.status}")
        self.status = OrderStatus.CONFIRMED

Маппер — в adapters/out/persistence/order_mapper.py:

# adapters/out/persistence/order_mapper.py
from core.order.entities import Order, OrderItem, OrderStatus
from adapters.out.persistence.models import OrderModel, OrderItemModel


def to_domain(model: OrderModel) -> Order:
    return Order(
        id=model.id,
        customer_id=model.customer_id,
        status=OrderStatus(model.status),
        total_amount=model.total_amount,
        created_at=model.created_at,
        items=[_item_to_domain(item) for item in model.items],
    )


def to_model(order: Order) -> OrderModel:
    model = OrderModel(
        id=order.id,
        customer_id=order.customer_id,
        status=order.status.value,
        total_amount=order.total_amount,
        created_at=order.created_at,
    )
    model.items = [_item_to_model(item) for item in order.items]
    return model


def _item_to_domain(model: OrderItemModel) -> OrderItem:
    return OrderItem(
        id=model.id,
        product_id=model.product_id,
        quantity=model.quantity,
        unit_price=model.unit_price,
    )


def _item_to_model(item: OrderItem) -> OrderItemModel:
    return OrderItemModel(
        id=item.id,
        product_id=item.product_id,
        quantity=item.quantity,
        unit_price=item.unit_price,
    )

to_domain и to_model — не методы, а функции. Они не хранят состояние и не нуждаются в классе. Если маппер требует внешних зависимостей (например, сервис конверсии валют или ObjectMapper для JSONB-колонки), тогда оправдан класс с инжектируемыми зависимостями.

Сборка агрегата — в маппере, не в репозитории

Правило R-SQLA-MAP-2: сборка агрегата из строк целиком в маппере, репозиторий только вызывает to_domain(model).

Репозиторий загружает OrderModel с eager-loaded items через selectinload:

# adapters/out/persistence/order_repository.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from core.order.entities import Order
from core.order.port import OrderRepository
from adapters.out.persistence.models import OrderModel
from adapters.out.persistence.order_mapper import to_domain, to_model


class SqlAlchemyOrderRepository(OrderRepository):

    async def get(self, session: AsyncSession, order_id: uuid.UUID) -> Order | None:
        stmt = (
            select(OrderModel)
            .where(OrderModel.id == order_id)
            .options(selectinload(OrderModel.items))
        )
        result = await session.execute(stmt)
        model = result.scalar_one_or_none()
        if model is None:
            return None
        return to_domain(model)

    async def save(self, session: AsyncSession, order: Order) -> None:
        model = to_model(order)
        await session.merge(model)

session.merge работает для insert + upsert. Для insert-only используется session.add(model). Детали — в python/session-transactions.md.

Маппинг до commit

R-SQLA-SESS-X2 запрещает использовать ORM-объект после commit в async-контексте: expire_on_commit=True (по умолчанию) помечает атрибуты устаревшими, обращение к ним вне сессии вызывает MissingGreenlet.

Решение — маппить в domain до выхода из транзакции:

async def handle(self, cmd: ConfirmOrder) -> uuid.UUID:
    async with self._session_factory() as session, session.begin():
        order = await self._orders.get(session, cmd.order_id)
        if order is None:
            raise OrderNotFound(cmd.order_id)
        order.confirm()
        await self._orders.save(session, order)
        return order.id

Или настроить AsyncSession с expire_on_commit=False:

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Оба подхода приемлемы. Маппинг-до-commit чище: он делает явным, что ORM-объект не переживает транзакцию.

Customer: JSONB-адрес как value object

Когда колонка хранит JSONB (например, delivery_address), маппер распаковывает её в value object:

# adapters/out/persistence/models.py
from sqlalchemy.dialects.postgresql import JSONB

class CustomerModel(Base):
    __tablename__ = "customers"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True)
    name: Mapped[str] = mapped_column(String(256), nullable=False)
    delivery_address: Mapped[dict] = mapped_column(JSONB, nullable=True)

# adapters/out/persistence/customer_mapper.py
from core.customer.entities import Customer, Address


def to_domain(model: CustomerModel) -> Customer:
    address = None
    if model.delivery_address:
        raw = model.delivery_address
        address = Address(
            city=raw["city"],
            street=raw["street"],
            zip_code=raw.get("zip_code"),
        )
    return Customer(
        id=model.id,
        name=model.name,
        delivery_address=address,
    )


def to_model(customer: Customer) -> CustomerModel:
    address_dict = None
    if customer.delivery_address:
        address_dict = {
            "city": customer.delivery_address.city,
            "street": customer.delivery_address.street,
            "zip_code": customer.delivery_address.zip_code,
        }
    return CustomerModel(
        id=customer.id,
        name=customer.name,
        delivery_address=address_dict,
    )

SQLAlchemy автоматически сериализует/десериализует dict в JSONB. Маппер отвечает только за трансформацию dict ↔ Address.

Product: enum-конверсия

Enum в БД хранится строкой. Маппер переводит явно — не через __dict__:

# core/product/entities.py
class ProductStatus(str, Enum):
    ACTIVE = "active"
    ARCHIVED = "archived"
    OUT_OF_STOCK = "out_of_stock"


# adapters/out/persistence/product_mapper.py
def to_domain(model: ProductModel) -> Product:
    return Product(
        id=model.id,
        name=model.name,
        status=ProductStatus(model.status),
        price=model.price,
    )

def to_model(product: Product) -> ProductModel:
    return ProductModel(
        id=product.id,
        name=product.name,
        status=product.status.value,
        price=product.price,
    )

Если model.status содержит значение, отсутствующее в ProductStatus, ProductStatus(model.status) бросит ValueError — сразу, на слое persistence. Это правильное поведение: схема БД и доменная модель разошлись, и это ошибка, которую нужно фиксировать явно.

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

АнтипаттернПравилоЧто взамен
OrderModel.__dict__ или vars(model) для конвертации в domainR-SQLA-MAP-X1Явная функция to_domain(model: OrderModel) -> Order
Возврат OrderModel из публичного метода репозиторияR-SQLA-REPO-X1Маппинг до выхода из репозитория, наружу — только Order
Бизнес-логика в маппере (if model.status == "cancelled": ...)R-SQLA-MODEL-X1Логика в доменных методах; маппер — только структурная конвертация
Сборка агрегата размазана по методам репозиторияR-SQLA-MAP-2Один to_domain(model) собирает агрегат целиком
lazy="select" на relationship в asyncR-SQLA-QRY-X2lazy="raise" по умолчанию + явный selectinload в запросе
Доступ к атрибутам ORM-модели после commit без expire_on_commit=FalseR-SQLA-SESS-X2Маппинг в domain до выхода из транзакции

Куда дальше

  • python/repository-pattern.md — как организован SqlAlchemyOrderRepository, порт в core/, инжекция AsyncSession.
  • python/session-transactions.md — Unit of Work, граница транзакции на handler'е, async with session.begin().
  • Маппинг record ↔ domain в jOOQ — Java-аналог: assembleAggregate, plain Java @Component, JSONB через JooqJsonbHelper; архитектурные решения те же, инструменты разные.