Aggregates
Clúster de objetos relacionados tratado como una unidad de consistencia. La raíz es la única puerta de entrada.
Un Aggregate es un grupo de objetos del dominio (Entities y Value Objects) que forman una unidad de consistencia. Todos los invariantes que involucran a los objetos del clúster se verifican dentro de este grupo. Desde el exterior, el Aggregate es opaco: solo se puede interactuar con él a través de su Aggregate Root.
El Aggregate Root es la Entity principal del clúster. Es la única referencia que el mundo exterior puede tener: repositorios guardan y recuperan Aggregates completos por el id de la raíz, y los application services solo interactúan con la raíz. Las Entities internas no tienen repositorio propio.
La regla más importante de los Aggregates es la de la referencia por id: si un Aggregate necesita relacionarse con otro Aggregate, lo hace almacenando el id (no una referencia al objeto). Esto mantiene los límites claros, evita cargar objetos innecesarios y hace que cada Aggregate pueda vivir en su propia transacción.
El tamaño de un Aggregate es una decisión de diseño crítica. La tendencia natural es crear Aggregates grandes ('Dios Aggregate') que lo incluyen todo. La regla práctica es: si no hay invariante que justifique que dos objetos estén en el mismo clúster, deben ser Aggregates separados. Los Aggregates pequeños son más fáciles de cambiar, menos propensos a conflictos de concurrencia y más eficientes.
Un Aggregate define un límite de transaccional: una transacción debe modificar un solo Aggregate. Si necesitas cambiar dos Aggregates en la misma operación de negocio, es señal de que o bien están en el mismo Aggregate (si comparten invariantes), o bien debes usar Domain Events y consistencia eventual para coordinarlos.
Los Domain Events son el mecanismo por el que un Aggregate notifica al mundo que algo relevante ocurrió dentro de él. El Aggregate registra los eventos en una lista interna; la capa de aplicación (o el Unit of Work) los despacha al Event Bus después de confirmar la transacción. El Aggregate nunca publica directamente al bus.
# --- shared_kernel/domain/aggregate_root.py ---
# Base Aggregate Root: registra Domain Events para despacho post-transacción.
# NO despacha eventos directamente; eso es responsabilidad de la capa de aplicació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:
# Acumula eventos; se despachan DESPUÉS de confirmar la transacción
self._domain_events.append(event)
def pull_events(self) -> list[DomainEvent]:
# La capa de aplicación llama esto para obtener y limpiar los eventos
events = list(self._domain_events)
self._domain_events.clear()
return events
# --- contexts/<bc>/domain/model/<aggregate>.py ---
# Aggregate Root del bounded context. Única puerta de entrada al clúster.
# Protege invariantes; referencia otros Aggregates SOLO POR ID.
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from <project>.shared_kernel.domain.aggregate_root import AggregateRoot
from <project>.contexts.<bc>.domain.model.<value_object> import <ValueObject>
from <project>.contexts.<bc>.domain.model.<entity> import <Entity>
from <project>.contexts.<bc>.domain.events.<something_happened> import <SomethingHappened>
@dataclass
class <Aggregate>(AggregateRoot):
id: UUID = field(default_factory=uuid4)
<attribute>: <ValueObject> = ...
# Referencia a otro Aggregate POR ID (no por objeto)
<other_aggregate_id>: UUID = ...
# Entities internas del clúster
_<items>: list[<Entity>] = field(default_factory=list, repr=False)
def <domain_action>(self, <param>: <ValueObject>) -> None:
# 1. Verificar pre-condición / invariante
self._ensure_<invariant>()
# 2. Aplicar cambio de estado
self.<attribute> = <param>
# 3. Registrar Domain Event (no despachar)
self._register_event(<SomethingHappened>(aggregate_id=self.id))
def _ensure_<invariant>(self) -> None:
if not <condition>:
raise <DomainError>("<invariant> violated")
# WRONG — referencia por objeto: acopla los Aggregates y carga innecesario
# other_aggregate: OtherAggregate = ... # ← NO; usar OtherAggregateId: UUIDDebugging lab
Detecta y corrige el error o la violación de diseño.
- 3.3.5.1
@dataclass class <Aggregate>(AggregateRoot): id: UUID owner: 'UserAggregate' # referencia al objeto completo
- 3.3.5.2
def complete_order_and_update_stock(order_id: UUID, product_id: UUID) -> None: order = order_repo.get_by_id(order_id) stock = stock_repo.get_by_id(product_id) order.complete() stock.decrement(1) session.commit() # 1 transacción, 2 Aggregates
- 3.3.5.3
# Repositorio de Entity interna class LineItemRepository: def get_by_id(self, line_item_id: UUID) -> LineItem: ...
- 3.3.5.4
class <Aggregate>(AggregateRoot): def <domain_action>(self) -> None: self.status = 'done' event_bus.publish(<SomethingHappened>(self.id)) # publicar dentro del Aggregate
- 3.3.5.5
@dataclass class <Aggregate>(AggregateRoot): id: UUID items: list['Item'] related_config: 'ConfigAggregate' # otro aggregate user_profile: 'ProfileAggregate' # otro aggregate audit_log: list['AuditEntry'] # historial de auditoría