DDD Tactical: Entity, Value Object, Aggregate

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

Tactical DDD — конкретный инструмент проектирования domain-модели. Без него Clean Architecture и Hexagonal превращаются в пустую скорлупу: слои есть, но что в них лежит — непонятно. Entity, Value Object и особенно Aggregate — три понятия, которые определяют, как думать о бизнес-логике.

Ошибка в границах aggregate — самая дорогая ошибка в DDD. Разберём её отдельно.

Entity vs Value Object

Разница не в структуре, а в семантике:

• Два Customer с одинаковыми именами и адресами — разные люди. Это Entity: важна идентичность.
• Два Money(100, 'RUB') — одно и то же. Это Value Object: важны значения.

Тест: если поменять поле, это ещё тот же объект?

• Customer.email = 'new@example.com' — тот же клиент. Entity.
• Money.amount = 200 — уже другая сумма. Value Object.

Пример Entity

@dataclass
class Order:
    id: OrderId                            
    customer_id: CustomerId
    items: list[OrderLine]
    status: OrderStatus
    placed_at: datetime

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Order):
            return NotImplemented
        return self.id == other.id           

    def __hash__(self) -> int:
        return hash(self.id)

• Идентичность — id (obычно OrderId — сам является Value Object).
• Равенство — только по id, независимо от полей.
• status может меняться со временем — заказ остаётся тем же.

Пример Value Object

@dataclass(frozen=True)                    
class Money:
    amount: Decimal
    currency: Currency

    def __post_init__(self):
        if self.amount < 0:
            raise DomainError('amount must be non-negative')

    def __add__(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise DomainError(f'cannot add {self.currency} and {other.currency}')
        return Money(self.amount + other.amount, self.currency)

    def __mul__(self, factor: Decimal) -> 'Money':
        return Money(self.amount * factor, self.currency)

• frozen=True — immutable. Гарантия на уровне рантайма.
• Инкапсулирует правила — нельзя сложить рубли с долларами, нельзя быть отрицательным.
• Равенство автоматом — через @dataclass по полям.
• Operator overloading — money + money, money * factor работают как ожидается.

Aggregate

Aggregate — самое сложное понятие в DDD и самое дорогое, если ошибиться.

Пример: Order + OrderLine

@dataclass
class OrderLine:                                  
    product_id: ProductId
    quantity: int
    unit_price: Money

    @property
    def total(self) -> Money:
        return self.unit_price * self.quantity


@dataclass
class Order:                                      
    """Aggregate Root."""
    id: OrderId
    customer_id: CustomerId
    _lines: list[OrderLine]
    status: OrderStatus = OrderStatus.DRAFT

    @property
    def total(self) -> Money:
        return sum((line.total for line in self._lines), start=Money.zero('RUB'))

    def add_line(self, product_id: ProductId, quantity: int, price: Money) -> None:
        if self.status is not OrderStatus.DRAFT:
            raise DomainError('cannot add lines to non-draft order')
        if quantity <= 0:
            raise DomainError('quantity must be positive')
        self._lines.append(OrderLine(product_id, quantity, price))

    def submit(self) -> None:
        if self.status is not OrderStatus.DRAFT:
            raise DomainError(f'cannot submit order in state {self.status}')
        if not self._lines:
            raise DomainError('cannot submit empty order')
        self.status = OrderStatus.SUBMITTED

Что здесь важно:

• Order — Root, OrderLine — child. Внешний код никогда не работает с OrderLine напрямую.
• _lines — private. Модификации только через методы Order.
• Инвариант защищён. «Нельзя добавить строку в submitted заказ» проверяется внутри add_line. Никакой внешний код не может сломать это правило.
• total вычисляется, а не хранится. Не может рассинхронизироваться с реальностью.

Почему нельзя работать с OrderLine напрямую

Ситуация без Root:

# где-то в use case
order = orders.find(order_id)
line = order.lines[0]
line.quantity = 5             # ← прямая модификация child'а     

Проблемы:

1. Инвариант Order не проверен. Может быть, заказ уже submitted, а вы поменяли количество.
2. Order.total не пересчитан — если это кэшированное поле, оно устарело.
3. Логика размазана. «Как правильно менять quantity» разбросано по use case’ам.

Правильно:

order.change_quantity(product_id, 5)   # метод aggregate root

Внутри проверяется всё, что нужно. Одна точка изменения — одна точка контроля.

Правила Vernon для Aggregate

В Effective Aggregate Design (2011) Vaughn Vernon сформулировал 4 правила, которые сохраняют aggregate работоспособными.

Правило 1. Модель настоящие инварианты внутри aggregate.

Только правила, которые должны быть консистентны транзакционно, живут внутри aggregate. Всё остальное — вне.

Пример: «сумма order.total = сумме line.total» — инвариант Order. Живёт внутри. «Клиент должен иметь лимит выше total» — инвариант между Order и Customer. Живёт в domain service или eventual consistency.

Правило 2. Проектируйте маленькие агрегаты.

Крупные aggregate — узкое место. Load всего заказа со всеми ~1000 lines в память для изменения одного поля — плохо. Крупный aggregate = крупный lock.

Vernon: «Small aggregates, please».

Правило 3. Ссылайтесь на другие aggregate по идентификатору.

Order.customer — не ссылка на объект Customer, а customer_id: CustomerId. Иначе загрузка Order тянет за собой Customer, Customer тянет свой AddressBook, и получается граф на пол-БД.

Правило 4. Используйте eventual consistency между aggregate.

Order.place() изменяет Order. Если нужно уменьшить Inventory.stock — это отдельный aggregate, отдельная транзакция. Между ними — событие (OrderPlaced) и обработчик (InventoryReserver).

Не пытайтесь быть strongly consistent между aggregate — вы боретесь с распределённой системой.

Пример правил 3 и 4

@dataclass
class Order:
    id: OrderId
    customer_id: CustomerId          # ← не Customer, а CustomerId              
    lines: list[OrderLine]
    total: Money

    def place(self, customer_credit_limit: Money) -> 'OrderPlaced':
        if self.total > customer_credit_limit:                                 
            raise DomainError('over credit limit')
        self.status = OrderStatus.PLACED
        return OrderPlaced(order_id=self.id, customer_id=self.customer_id)

• Ссылка на клиента — customer_id, не сам объект.
• Инвариант «total ≤ credit_limit» проверяется, но credit_limit передаётся снаружи — application service его загрузил и передал. Между Order и Customer нет прямой связи в памяти.
• place() возвращает событие. Реакция на него (напр. InventoryReserver) — в отдельной транзакции.

Domain Service

Иногда логика не помещается ни в одну Entity. Пример: правило начисления скидки, зависящее от Order, Customer и Region — трёх разных aggregate.

Признаки Domain Service:

• Оперирует несколькими aggregate.
• Нет собственного состояния (stateless).
• Логика чисто доменная — никакой инфраструктуры, никаких портов.
• Отдельный класс, но живёт в domain/.

Пример: PricingPolicy

class PricingPolicy:
    """Правило расчёта финальной цены заказа с учётом клиента и региона."""

    def calculate(
        self,
        order: Order,
        customer: Customer,
        region: Region,
    ) -> Money:
        base = order.subtotal
        discount = customer.discount_for(base)
        tax = region.apply_taxes(base - discount)
        return tax

• Оперирует тремя aggregate root’ами. Ни одной не «принадлежит».
• Нет состояния — можно вызвать как функцию.
• Инжектится в application service:

class PlaceOrderCommandHandler:
    def __init__(self, ..., pricing: PricingPolicy):
        self._pricing = pricing

    async def execute(self, cmd):
        order = ...
        customer = ...
        region = ...
        final = self._pricing.calculate(order, customer, region)
        order.apply_final_price(final)

Как не надо

1. Большие агрегаты

class Customer:
    orders: list[Order]                
    addresses: list[Address]           
    payment_methods: list[PaymentMethod]

    def add_order(self, order): ...
    def cancel_order(self, order_id): ...
    def add_address(self, address): ...
    def add_payment_method(self, pm): ...

Customer знает про все свои заказы. Загрузка Customer’а тянет все заказы. Транзакция создания нового заказа блокирует чтение старых. Ужас.

Правильно: Order — отдельный aggregate. Customer.orders — не поле, а запрос через repository: orders.find_by_customer(customer_id).

2. Прямые ссылки между aggregate

class Order:
    customer: Customer               

    def total_with_discount(self):
        return self.total - self.customer.discount   

Order держит объект Customer в памяти. При загрузке заказа — обязательно загрузить клиента. Если клиент удалён — orphan reference.

Правильно: customer_id: CustomerId + передавать Customer в методы через параметр (когда нужен) или через application service.

3. Setter’ы вместо методов

class Order:
    status: OrderStatus              # public!            
    total: Money                     # public!            

# внешний код:
order.status = OrderStatus.CONFIRMED    # !!             
order.total = new_total

Order анемичен. Инварианты не защищены. status = CONFIRMED можно установить, не проверив _lines.

Правильно: private-поля, методы для изменений. order.confirm(), order.recalculate_total().

4. Domain Service с состоянием или инфраструктурой

from app.infrastructure.tax_api_client import TaxApiClient   # !!  

class PricingPolicy:
    def __init__(self, client: TaxApiClient):                
        self._client = client
        self._cache: dict = {}                                

    def calculate(self, order, customer, region):
        tax = self._client.get_tax(region)
        ...

Domain Service держит state (_cache) и знает про инфраструктуру (TaxApiClient). Это уже не Domain Service, а Application Service.

Правильно: PricingPolicy — pure function. Данные (tax rates) передаются в параметрах. Кэширование — задача другого слоя.

Порядок работы в use case

class PlaceOrderCommandHandler:
    def __init__(
        self,
        orders: OrderRepositoryPort,
        customers: CustomerRepositoryPort,
        regions: RegionRepositoryPort,
        pricing: PricingPolicy,
        uow: UnitOfWork,
        events: EventPublisherPort,
    ) -> None:
        self._orders = orders
        self._customers = customers
        self._regions = regions
        self._pricing = pricing
        self._uow = uow
        self._events = events

    async def execute(self, cmd: PlaceOrderCommand) -> OrderId:
        async with self._uow:
            # 1. Загружаем нужные aggregate root'ы
            customer = await self._customers.find_by_id(cmd.customer_id)
            if customer is None:
                raise CustomerNotFound(cmd.customer_id)
            region = await self._regions.find_by_id(customer.region_id)

            # 2. Создаём новый aggregate
            order = Order.draft(customer_id=customer.id)
            for item in cmd.items:
                order.add_line(item.product_id, item.quantity, item.price)

            # 3. Применяем domain service
            final_price = self._pricing.calculate(order, customer, region)
            order.apply_final_price(final_price)

            # 4. Совершаем бизнес-действие
            event = order.place(customer.credit_limit)

            # 5. Персистим
            await self._orders.save(order)

            # 6. Публикуем событие
            await self._events.publish(event)

            return order.id

Шаги: load → create/mutate → save → publish. Логика — внутри aggregate и domain service. Application service — тонкий оркестратор.

Trade-offs

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

Анонимизированный микросервис почти не имеет domain-модели — папка domain/ пуста. Это приемлемо, потому что сервис-оркестратор: получил сообщение, распарсил, вызвал marketplace, сохранил, отправил. Собственных инвариантов нет.

Как бы это выглядело, если бы инварианты появились:

@dataclass
class NormalizationBatch:
    """Aggregate: пакет характеристик к нормализации."""
    id: BatchId
    item_id: ItemId
    _properties: list[PropertyToNormalize]
    _max_size: ClassVar[int] = 100

    def add_property(self, name: str, value: str, unit: str | None) -> None:
        if len(self._properties) >= self._max_size:
            raise DomainError(f'batch cannot exceed {self._max_size} properties')
        if not name:
            raise DomainError('property name is required')
        self._properties.append(PropertyToNormalize(name, value, unit))

    def can_be_normalized(self) -> bool:
        return len(self._properties) > 0

    def to_request_lines(self) -> list[str]:
        return [p.to_line() for p in self._properties]

Такой aggregate защитит инварианты «не больше 100 характеристик за раз», «не нормализуем пустой пакет». Сейчас эти проверки размазаны по _collect_properties в use case.

Правильно ли вводить aggregate? Только если инварианты действительно есть. Иначе — overengineering.

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

• Eric Evans. Domain-Driven Design. Part II. Entity, Value Object, Aggregate, Domain Service, Repository, Factory.
• Vaughn Vernon. Implementing Domain-Driven Design. Chapter 5–7 — Entity, VO, Aggregate.
• Vaughn Vernon. Effective Aggregate Design. 3 статьи. Обязательно.
• Scott Wlaschin. Domain Modeling Made Functional. DDD через типы в F#. Идеи применимы к любому языку с типами.
• Harry Percival, Bob Gregory. Architecture Patterns with Python. Chapter 7 — Aggregates и consistency boundaries.

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