Опирается на правила: R-AGG-1 … R-AGG-5 и R-AGG-X1 … R-AGG-X4 из DDD Tactical Style Guide → раздел 3. Aggregate Root.

Важно знать

  • Агрегат — кластер сущностей и VO, согласованных по инвариантам. Единственная точка внешнего доступа — корень AggregateRoot.
  • Транзакционная граница = граница агрегата. Один use-case меняет один агрегат. Между агрегатами — eventual consistency через доменные события.
  • Внутренние Entity скрыты. Коллекции наружу — только через tuple(self._lines) или срез, никогда return self._lines.
  • Доменные события регистрируются только внутри методов корня через self._register_event(...). Репозитории, Handler-ы и контроллеры события не регистрируют.
  • Ссылки на другие агрегаты — по ID (CustomerId, ProductId), не объекты.
  • Корень выделяется по бизнес-инварианту, который нужно поддерживать атомарно. Не «всё, что связано с заказом», а «то, что нельзя обновить по отдельности».
  • God aggregate (десятки несвязанных Entity внутри) — главный антипаттерн. Делить по инвариантам, не по UI и не по схеме БД.

Aggregate Root — объект, отвечающий за согласованность кластера данных. Если правило «сумма позиций заказа всегда равна total» — Order агрегат, OrderLine внутренняя сущность, и любая операция над позицией идёт через метод Order. Если бы OrderLine был самостоятельным агрегатом, инвариант пришлось бы поддерживать снаружи — двухфазными коммитами или eventual consistency. Раскрытие раздела 3 гайда.

Базовый класс AggregateRoot

R-AGG-1: корень агрегата наследует AggregateRoot[ID] из core/shared/building_blocks.py. Это надстройка над Entity[ID], которая добавляет список накопленных событий и два метода: _register_event (защищённый, для методов самого корня) и pull_events (для репозитория/UoW).

# core/shared/building_blocks.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import Generic, TypeVar
from uuid import UUID

ID = TypeVar("ID")


class Entity(Generic[ID]):
    def __init__(self, id_: ID) -> None:
        self._id = id_

    @property
    def id(self) -> ID:
        return self._id

    def __eq__(self, other: object) -> bool:
        return isinstance(other, type(self)) and self._id == other._id

    def __hash__(self) -> int:
        return hash((type(self).__name__, self._id))


@dataclass(frozen=True)
class DomainEvent:
    event_id: UUID
    occurred_at: datetime
    aggregate_id: UUID


class AggregateRoot(Entity[ID]):
    def __init__(self, id_: ID) -> None:
        super().__init__(id_)
        self._events: list[DomainEvent] = []

    def _register_event(self, event: DomainEvent) -> None:
        self._events.append(event)

    def pull_events(self) -> list[DomainEvent]:
        events = list(self._events)
        self._events.clear()
        return events

Все внешние операции — через методы корня

R-AGG-2: внутренние Entity недоступны снаружи без обёртки. Любая модификация состояния — через метод корня. Коллекции наружу — через tuple(self._lines).

# core/order/aggregate/order.py
from uuid import UUID, uuid4
from core.shared.building_blocks import AggregateRoot
from core.order.value_object.order_id import OrderId
from core.order.value_object.customer_id import CustomerId
from core.order.value_object.money import Money
from core.order.value_object.order_status import OrderStatus
from core.order.entity.order_line import OrderLine
from core.order.event.order_confirmed import OrderConfirmed
from core.order.event.order_line_added import OrderLineAdded
from core.shared.clock import Clock


class Order(AggregateRoot[OrderId]):
    def __init__(self, id_: OrderId, customer_id: CustomerId) -> None:
        super().__init__(id_)
        self._customer_id = customer_id
        self._status = OrderStatus.NEW
        self._lines: list[OrderLine] = []

    @property
    def status(self) -> OrderStatus:
        return self._status

    @property
    def lines(self) -> tuple[OrderLine, ...]:
        return tuple(self._lines)

    def add_line(self, line: OrderLine) -> None:
        if self._status != OrderStatus.NEW:
            raise ValueError(f"cannot add line to order in status {self._status}")
        self._lines.append(line)
        self._register_event(
            OrderLineAdded(uuid4(), clock_now(), self.id.value, line.id.value)
        )

    def confirm(self, clock: Clock) -> None:
        if not self._lines:
            raise ValueError("cannot confirm empty order")
        if self._status != OrderStatus.NEW:
            raise ValueError(f"order already in status {self._status}")
        self._status = OrderStatus.CONFIRMED
        self._register_event(
            OrderConfirmed(
                uuid4(),
                clock.now(),
                self.id.value,
                customer_id=self._customer_id.value,
                total=self._total().amount,
                currency=self._total().currency,
            )
        )

    def _total(self) -> Money:
        from decimal import Decimal
        total = Decimal(0)
        for line in self._lines:
            total += line.subtotal().amount
        return Money(total, self._lines[0].subtotal().currency) if self._lines else Money(Decimal(0), "RUB")

Что неправильно:

# ПЛОХО — клиент мутирует внутренний список
order.lines.append(line)       # tuple не мутируется, но...
order._lines.append(line)      # прямой доступ к _lines — нарушение R-AGG-2

tuple(self._lines) даёт иммутабельный снимок. Клиент видит состояние агрегата, но не может его изменить в обход методов.

События регистрируются только в корне

R-AGG-3, R-AGG-X4: _register_event(...) вызывается внутри методов агрегата в момент изменения состояния. Не в Handler, не в репозитории, не в контроллере.

def cancel(self, clock: Clock, reason: str) -> None:
    if self._status == OrderStatus.SHIPPED:
        raise ValueError("cannot cancel shipped order")
    if self._status == OrderStatus.CANCELLED:
        return
    self._status = OrderStatus.CANCELLED
    self._register_event(
        OrderCancelled(uuid4(), clock.now(), self.id.value, reason=reason)
    )

Почему именно в корне:

  • Атомарность. Состояние и событие меняются в одном месте — невозможно изменить статус и «забыть» зарегистрировать событие.
  • Корректность. Только сам агрегат знает, действительно ли он перешёл в новое состояние. Идемпотентный early-return выше не регистрирует дублирующее событие.
  • Граница ответственности. Handler оркеструет, агрегат принимает решения.

Транзакционная граница — один агрегат

R-AGG-4: один use-case изменяет один агрегат. Другие агрегаты — только через события (eventual consistency).

# core/order/usecase/command/confirm_order.py
from dataclasses import dataclass
from core.order.value_object.order_id import OrderId


@dataclass(frozen=True)
class ConfirmOrder:
    order_id: OrderId


# core/order/usecase/command/confirm_order_handler.py
class ConfirmOrderHandler:
    def __init__(self, orders: OrderRepository, clock: Clock) -> None:
        self._orders = orders
        self._clock = clock

    async def handle(self, cmd: ConfirmOrder) -> None:
        order = await self._orders.by_id(cmd.order_id)
        if order is None:
            raise OrderNotFoundError(cmd.order_id)
        order.confirm(self._clock)          # меняет один агрегат
        await self._orders.save(order)      # репозиторий публикует события

OrderConfirmed затем обрабатывается отдельным процессом — например, резервирует склад или создаёт счёт. Это другие транзакции, в которых меняются их агрегаты.

Что нельзя:

# ПЛОХО — handler меняет два агрегата в одной транзакции
async def handle(self, cmd: ConfirmOrder) -> None:
    order = await self._orders.by_id(cmd.order_id)
    customer = await self._customers.by_id(order._customer_id)  # R-AGG-X3
    customer.increment_order_count()
    order.confirm(self._clock)
    await self._customers.save(customer)
    await self._orders.save(order)

R-AGG-X3: изменение чужого агрегата напрямую — deadlock-prone, нарушает локальность инвариантов, мешает выносить BC в отдельный сервис.

Ссылки между агрегатами — только по ID

R-AGG-5: внутри Order не хранится объект Customer. Только CustomerId.

# ПЛОХО
class Order(AggregateRoot[OrderId]):
    def __init__(self, id_: OrderId, customer: Customer) -> None:  # объект другого агрегата
        ...

# ХОРОШО
class Order(AggregateRoot[OrderId]):
    def __init__(self, id_: OrderId, customer_id: CustomerId) -> None:
        ...

Когда нужна «Order + имя клиента» на read-side — это read-model (CQRS), а не объект Customer внутри Order.

God aggregate — главный антипаттерн

R-AGG-X1: признак — внутри одного агрегата несвязанные Entity с разными жизненными циклами.

# ПЛОХО — God aggregate
class Customer(AggregateRoot[CustomerId]):
    def __init__(self, ...) -> None:
        ...
        self._orders: list[Order] = []          # каждый Order — свой агрегат
        self._invoices: list[Invoice] = []      # Invoice тоже агрегат
        self._subscriptions: list[Subscription] = []

При загрузке такого агрегата поднимаются мегабайты данных. Любая операция берёт блокировку на весь кластер. Транзакции конфликтуют.

Правильное деление:

  • Customer — id, имя, email, статус. Один агрегат.
  • Order — id, lines, status. Другой агрегат, хранит customer_id: CustomerId.
  • Invoice, Subscription — отдельные агрегаты со своими корнями.

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

АнтипаттернПравилоЧто взамен
God aggregate (несвязанные Entity внутри)R-AGG-X1Делить по локальному инварианту, ссылки по ID
return self._lines без обёрткиR-AGG-X2tuple(self._lines) или срез
Изменение чужого агрегата напрямуюR-AGG-X3_register_event + подписчик в отдельной транзакции
_register_event в Handler/репозиторииR-AGG-X4Только внутри методов самого агрегата
Объект другого агрегата как поле (customer: Customer)R-AGG-5customer_id: CustomerId

Куда дальше

  • DDD Tactical → раздел 3. Aggregate Root — нормативные формулировки R-AGG-*.
  • python/entity.md — внутренние Entity агрегата (не корни).
  • python/domain-event.md — что такое _register_event и как события публикуются после save.
  • python/repository.md — как агрегат сохраняется и поднимается через Protocol-порт.
  • python/value-object.md — VO и ID-типы, которые ссылаются между агрегатами.
  • python/module-structure.md — где живут агрегаты в структуре core/.