Aggregate Root e Invariantes
La raíz protege la consistencia del clúster. Un Aggregate = una transacción. Nadie más puede romper las reglas.
Un invariante de dominio es una regla de negocio que debe ser verdadera en todo momento dentro de un Aggregate. 'Una orden no puede tener más de N líneas', 'el saldo de una cuenta no puede ser negativo', 'un pedido completado no puede ser cancelado' son invariantes. La responsabilidad de que estas reglas nunca se rompan es exclusiva del Aggregate Root.
La primera consecuencia de esta responsabilidad es que el Aggregate Root es la única puerta de entrada al clúster. Ningún objeto externo puede modificar directamente una Entity o Value Object interno. Toda modificación pasa por un método del Aggregate Root, que valida los invariantes antes y/o después del cambio.
La segunda consecuencia es la regla 'un Aggregate = una transacción'. Si una operación de negocio requiere modificar dos Aggregates, eso es señal de que o bien forman el mismo Aggregate (si comparten invariante), o bien deben coordinarse con consistencia eventual a través de Domain Events. No existe la opción de una transacción que abarque dos Aggregates.
Los invariantes deben ser verificados en el modelo de dominio, no en la capa de aplicación ni en la infraestructura. Si la validación está en el application service, es bypass-able: cualquier otro caller puede olvidar llamarla. Si está en el Aggregate Root, es imposible de saltarse porque es el único acceso.
El Aggregate Root también es el origen de los Domain Events. Cuando una operación de negocio tiene éxito (invariante verificada, estado cambiado), el Aggregate Root registra el evento correspondiente. Los eventos se despachan después de que la transacción sea confirmada, garantizando que nunca se publiquen eventos de operaciones que finalmente fallaron.
Diseñar el límite del Aggregate correctamente es uno de los desafíos más importantes de DDD. El principio guía es: incluye en un Aggregate solo los objetos cuya consistencia conjunta debe garantizarse en la misma transacción. Si puedes relajar la consistencia entre dos conceptos y aceptar eventual consistency, mantenlos como Aggregates separados.
# --- shared_kernel/domain/aggregate_root.py ---
# Base Aggregate Root. Los Aggregates concretos heredan de esta clase.
# NO despacha eventos; solo los registra para despacho post-transacción.
from <project>.shared_kernel.domain.entity import Entity
from <project>.shared_kernel.domain.domain_event import DomainEvent
class AggregateRoot(Entity):
def __init__(self) -> None:
self._domain_events: list[DomainEvent] = []
def _register_event(self, event: DomainEvent) -> None:
self._domain_events.append(event)
def pull_events(self) -> list[DomainEvent]:
events = list(self._domain_events)
self._domain_events.clear()
return events
# --- contexts/<bc>/domain/model/<aggregate>.py ---
# Protege invariantes. Única puerta de entrada al clúster.
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from enum import Enum
from <project>.shared_kernel.domain.aggregate_root import AggregateRoot
from <project>.shared_kernel.domain.errors import DomainError
from <project>.contexts.<bc>.domain.model.<value_object> import <ValueObject>
from <project>.contexts.<bc>.domain.events.<something_happened> import <SomethingHappened>
class <AggregateStatus>(str, Enum):
DRAFT = 'draft'
ACTIVE = 'active'
CLOSED = 'closed'
@dataclass
class <Aggregate>(AggregateRoot):
id: UUID = field(default_factory=uuid4)
status: <AggregateStatus> = <AggregateStatus>.DRAFT
<attribute>: <ValueObject> = ...
_<items>: list[<Item>] = field(default_factory=list, repr=False)
# Invariante 1: el Aggregate solo puede activarse desde DRAFT
def activate(self) -> None:
if self.status != <AggregateStatus>.DRAFT:
raise DomainError(f"Cannot activate: status is {self.status}")
self.status = <AggregateStatus>.ACTIVE
self._register_event(<SomethingHappened>(aggregate_id=self.id))
# Invariante 2: no se puede añadir un item a un Aggregate cerrado
def add_<item>(self, item: <Item>) -> None:
if self.status == <AggregateStatus>.CLOSED:
raise DomainError("Cannot add item: aggregate is closed")
# Invariante 3: unicidad dentro del clúster
if any(i.id == item.id for i in self._<items>):
raise DomainError("Item already exists in this aggregate")
self._<items>.append(item)
# Invariante 4: solo la raíz puede cerrar, y solo si hay items
def close(self) -> None:
if not self._<items>:
raise DomainError("Cannot close: no items")
self.status = <AggregateStatus>.CLOSED
self._register_event(<SomethingHappened>(aggregate_id=self.id))
# WRONG — validar invariante FUERA del Aggregate:
# En el application service:
# if aggregate.status != 'draft': # ← cualquiera puede olvidar esta validación
# raise ValueError('...')
# aggregate.status = 'active' # ← mutación directa, sin pasar por el métodoDebugging lab
Detecta y corrige el error o la violación de diseño.
- 3.4.5.1
# En el application service: def activate_<aggregate>(agg_id: UUID) -> None: agg = repo.get_by_id(agg_id) if agg.status != 'draft': raise ValueError('must be draft') # validación fuera agg.status = 'active' # mutación directa repo.save(agg)
- 3.4.5.2
def complete_purchase(order_id: UUID, product_id: UUID) -> None: order = order_repo.get_by_id(order_id) inventory = inventory_repo.get_by_id(product_id) order.complete() inventory.decrement_stock(1) session.commit() # 2 Aggregates en 1 transacción
- 3.4.5.3
class <Aggregate>(AggregateRoot): def add_item(self, item: <Item>) -> None: self._items.append(item) # sin verificar invariantes
- 3.4.5.4
class <Aggregate>(AggregateRoot): def close(self) -> None: self.status = <AggregateStatus>.CLOSED import event_bus # importar infra dentro del dominio event_bus.publish(<AggregateClosed>(self.id)) # publicar directo
- 3.4.5.5
@dataclass class <Aggregate>(AggregateRoot): id: UUID status: str = 'active' def pull_events(self) -> list[DomainEvent]: # NO limpiar la lista después de retornar return list(self._domain_events)