Mappers и изоляция ORM

Зачем эта глава

Каждый второй DDD-туториал на Python выглядит так:

class Order(Base):
    __tablename__ = 'orders'
    id: Mapped[int] = mapped_column(primary_key=True)
    total: Mapped[Decimal]

    def confirm(self):
        self.status = 'confirmed'

Красиво. Работает. Тесты и запросы через ORM. Domain и infrastructure слились в одну сущность. Это Active Record — валидный паттерн, но не совместимый с чистым Hexagonal/Onion/Clean.

Разберём, что теряется, что выигрывается, и как правильно оформить mapping, если хотим держать domain чистым.

Active Record

Классический пример — Django ORM, Ruby on Rails, Ecto (Elixir):

class Order(Model):
    total = DecimalField()
    status = CharField()

    def confirm(self):
        self.status = 'confirmed'
        self.save()

# использование:
order = Order.objects.get(id=1)
order.confirm()

Плюсы:

• Быстрое написание. Один класс = вся сущность.
• Понятно новичкам.
• CRUD пишется в 3 строки.

Минусы:

• Domain-класс наследуется от ORM-класса. Правило зависимостей нарушено.
• Domain-инварианты защитить сложно — persistence логика их не проверяет.
• Тестировать без БД нельзя.
• Смена ORM = переписать всё.

Active Record хорош для CRUD-приложений с простым доменом. Для сложного домена и Clean Architecture — не подходит.

Data Mapper

Domain-класс — чистый Python:

from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum

class OrderStatus(Enum):
    DRAFT = 'draft'
    PLACED = 'placed'
    CONFIRMED = 'confirmed'

@dataclass
class Order:
    id: int | None                                # None до save
    customer_id: int
    total: Decimal
    status: OrderStatus = OrderStatus.DRAFT
    _events: list = field(default_factory=list)

    def confirm(self) -> None:
        if self.status is not OrderStatus.PLACED:
            raise DomainError(f'cannot confirm order in state {self.status}')
        self.status = OrderStatus.CONFIRMED
        self._events.append(OrderConfirmed(order_id=self.id))

ORM-модель — отдельный класс в infrastructure:

from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase

class Base(DeclarativeBase):
    pass

class OrderRow(Base):
    __tablename__ = 'orders'

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int]
    total: Mapped[Decimal]
    status: Mapped[str]

Mapper — тонкий:

def to_domain(row: OrderRow) -> Order:
    return Order(
        id=row.id,
        customer_id=row.customer_id,
        total=row.total,
        status=OrderStatus(row.status),
    )

def to_orm(order: Order) -> OrderRow:
    return OrderRow(
        id=order.id,
        customer_id=order.customer_id,
        total=order.total,
        status=order.status.value,
    )

Repository использует mapper:

class PostgresOrderRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def find_by_id(self, order_id: int) -> Order | None:
        row = await self._session.get(OrderRow, order_id)
        return to_domain(row) if row else None

    async def save(self, order: Order) -> None:
        row = to_orm(order)
        await self._session.merge(row)

Плюсы:

• Domain — чистый, тестируется без БД.
• Замена SQLAlchemy — правится только infrastructure слой.
• Инварианты домена защищены — они не могут быть нарушены через ORM.

Минусы:

• Дублирование: Order и OrderRow со схожими полями.
• Mapper нужно поддерживать.
• Больше кода.

SQLAlchemy Imperative Mapping

SQLAlchemy 2.x позволяет мапить plain-классы без наследования от Base. Компромисс: domain остаётся чистым, но появляется runtime-магия.

# Чистый Python, никакого SQLAlchemy
@dataclass
class Order:
    id: int | None
    customer_id: int
    total: Decimal
    status: OrderStatus = OrderStatus.DRAFT

from sqlalchemy import Table, Column, Integer, Numeric, String, MetaData
from sqlalchemy.orm import registry

from app.domain.order import Order

metadata = MetaData()
mapper_registry = registry(metadata=metadata)

orders_table = Table(
    'orders',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('customer_id', Integer, nullable=False),
    Column('total', Numeric, nullable=False),
    Column('status', String(20), nullable=False),
)

mapper_registry.map_imperatively(Order, orders_table)     

Теперь Order сохраняется через session.add(order), но domain-класс не знает про SQLAlchemy.

Плюсы:

• Domain чистый.
• Не надо писать mapper вручную.
• SQLAlchemy features работают (identity map, change tracking).

Минусы:

• Enum-конверсия и сложные типы требуют custom TypeDecorator.
• _events (для domain events) SQLAlchemy пытается смапить как поле — нужен exclude_properties.
• Дебажить тяжелее — mapping в другом файле.

Percival и Gregory в Architecture Patterns with Python активно используют imperative mapping — рекомендую посмотреть их пример.

Async lazy loading — граната

Одна из самых частых ошибок при переходе на AsyncSession — забыть про lazy loading.

class OrderRow(Base):
    __tablename__ = 'orders'
    id: Mapped[int] = mapped_column(primary_key=True)
    lines: Mapped[list['OrderLineRow']] = relationship()   

async def find_order(session, order_id):
    order = await session.get(OrderRow, order_id)
    for line in order.lines:                                # !! MissingGreenlet    
        ...

order.lines пытается лениво загрузить OrderLineRow. В async-контексте это невозможно — SQLAlchemy бросает MissingGreenlet.

Решение: eager loading через selectinload или joinedload.

from sqlalchemy.orm import selectinload

async def find_order(session, order_id):
    stmt = (
        select(OrderRow)
        .where(OrderRow.id == order_id)
        .options(selectinload(OrderRow.lines))     
    )
    result = await session.execute(stmt)
    order = result.scalar_one_or_none()

Правило для aggregate: repository всегда возвращает полностью загруженный aggregate root со всеми child entities. Клиент не должен думать про lazy loading — этой абстракции у него нет.

Как не надо

1. Domain наследуется от Base

from sqlalchemy.orm import Mapped, mapped_column                          # !!  
from app.infrastructure.db.base import Base                                # !!

class Order(Base):                                                         
    __tablename__ = 'orders'
    id: Mapped[int] = mapped_column(primary_key=True)
    total: Mapped[Decimal]

    def confirm(self):
        self.status = 'confirmed'

Domain знает про SQLAlchemy. Правило зависимостей нарушено. Тесты требуют настраивать AsyncSession. Смена ORM = переписать domain.

Правильно: Order — чистый Python. Mapping отдельно.

2. Repository возвращает ORM-модель

class OrderRepository:
    async def find_by_id(self, id) -> OrderRow:               # !!    
        row = await self._session.get(OrderRow, id)
        return row

# use case:
order = await orders.find_by_id(1)
order.status = 'confirmed'                                    # !!    

Use case работает с ORM-моделью. Меняет поле напрямую — обходя доменные инварианты. Order.confirm() (защита состояния) вообще нет.

Правильно: Repository возвращает domain Order, use case работает с ним.

3. Mapper в domain-слое

class Order:
    @classmethod
    def from_row(cls, row: OrderRow) -> 'Order':               # !!  
        return cls(id=row.id, total=row.total, ...)

    def to_row(self) -> OrderRow:                              # !!  
        return OrderRow(id=self.id, total=self.total, ...)

Order знает про OrderRow. Domain импортирует infrastructure. Правило зависимостей снова нарушено.

Правильно: mapper — отдельный модуль в infrastructure, знает про оба слоя, оба слоя не знают про него.

4. Lazy loading в async

Уже разобрали выше. Всегда selectinload/joinedload для child entities aggregate.

5. N+1 запрос в use case

async def execute(self, cmd):
    orders = await self._orders.find_all_for_customer(cmd.customer_id)
    for order in orders:
        customer = await self._customers.find_by_id(order.customer_id)   # !!  

Классический N+1. Один запрос на orders, N запросов на customers.

Правильно:

• Один запрос за customers by IDs.
• Или query object / specification, чтобы Repository мог оптимизировать.
• Или (лучше) — вообще не делать так: подгружать customer при первой загрузке, если это нужно use case’у.

Trade-offs

В твоём же коде

Анонимизированный микросервис использует Data Mapper неявно: SQLAlchemy Base-модель TenderPositionAttributes живёт в infrastructure/db/models.py, domain-модели нет (сервис-оркестратор).

Если бы domain появился, разумный путь — Imperative Mapping. Domain — plain dataclass, mapping в отдельном файле, никаких Mapped-полей в бизнес-коде.

Что важно для case study:

• Не поддавайтесь искушению добавить бизнес-логику в TenderPositionAttributes (ORM-модель). Инварианты — в domain-классе.
• Repository возвращает domain-объект, не Row.

Дальнейшее чтение

• Martin Fowler. Patterns of Enterprise Application Architecture. Chapter 10 — Data Mapper, Active Record, Table Data Gateway.
• SQLAlchemy Docs. Imperative Mapping. Официальная документация.
• Harry Percival, Bob Gregory. Architecture Patterns with Python. Chapter 3 — Repository + imperative mapping. Обязательное чтение для Python-разработчика.
• SQLAlchemy Docs. ORM Querying Guide — Loading Techniques. selectinload, joinedload.
• Vladimir Khorikov. Unit Testing. Chapter 8 — почему mapping слой критичен для тестируемости.

Проверьте себя