Domain Events
Hechos inmutables del pasado que comunican qué ocurrió dentro del dominio.
Un Domain Event representa algo significativo que ya ocurrió dentro del dominio. No es una intención ni una orden: es un hecho consumado. El nombre siempre va en pasado — <SomethingHappened> — y debe ser parte del lenguaje ubicuo del equipo.
Los Domain Events son estrictamente inmutables. Una vez creados, no se modifican. Llevan consigo todos los datos necesarios para que cualquier handler pueda reaccionar sin necesidad de consultar el aggregate de origen.
Los events se emiten desde el Aggregate Root al completar una operación con éxito. El aggregate los acumula internamente y la capa de aplicación los recolecta y despacha después de persistir el estado.
El contenido del event pertenece al dominio: IDs de dominio, Value Objects o escalares. Nunca lleva modelos ORM, objetos de sesión HTTP ni dependencias de infraestructura. Esta regla garantiza que el dominio sea portable y testeable de forma aislada.
Cuando múltiples bounded contexts necesitan reaccionar al mismo hecho, el Domain Event se convierte en el contrato de integración. Sin embargo, ese contrato debe ser cuidadosamente versionado; un cambio en la estructura del event puede romper consumidores lejanos.
En Python, la forma idiomática es usar `@dataclass(frozen=True)` para garantizar la inmutabilidad. El campo `occurred_on` de tipo `datetime` con valor por defecto al momento de creación es el mínimo indispensable.
# domain/shared_kernel/domain_event.py
# Base class: define el contrato mínimo de todo Domain Event.
# NO agregar lógica de negocio aquí.
from dataclasses import dataclass, field
from datetime import datetime, timezone
from uuid import UUID, uuid4
@dataclass(frozen=True)
class DomainEvent:
event_id: UUID = field(default_factory=uuid4) # Identidad única del event
occurred_on: datetime = field( # Cuándo ocurrió (UTC)
default_factory=lambda: datetime.now(timezone.utc)
)
# domain/events/<something_happened>.py
# Concrete Domain Event: nombre en past tense, inmutable, datos mínimos.
# NO importar módulos de infraestructura ni ORM.
from dataclasses import dataclass
from uuid import UUID
from .<shared_kernel>.domain_event import DomainEvent
@dataclass(frozen=True)
class <SomethingHappened>(DomainEvent):
<aggregate>_id: UUID # ID del aggregate que originó el evento
<relevant_data>: str # Datos necesarios para los handlers; solo escalares/VOs
# NO incluir el aggregate completo ni referencias a entidades mutables
# domain/model/<aggregate>.py (fragmento)
# El Aggregate Root acumula events; NO los despacha él mismo.
class <Aggregate>:
def __init__(self) -> None:
self._events: list[DomainEvent] = []
def <do_something>(self, <param>: <Type>) -> None:
# 1. Validar invariantes
# 2. Mutar estado
# 3. Registrar event — siempre al final
self._events.append(<SomethingHappened>(<aggregate>_id=self.id, ...))
def pull_events(self) -> list[DomainEvent]:
events, self._events = self._events, []
return events # El handler de aplicación los recolecta y despachaDebugging lab
Detecta y corrige el error o la violación de diseño.
- 4.1.5.1
class OrderPlaced(DomainEvent): order_id: UUID def place(self): self.status = 'placed' # ← error aquí
- 4.1.5.2
@dataclass(frozen=True) class ItemRemoved(DomainEvent): occurred_on: datetime = field(default_factory=datetime.now) # ← error aquí
- 4.1.5.3
@dataclass(frozen=True) class UserRegistered(DomainEvent): user: User # ← error aquí (User es una entidad mutable) session: Session # ← error aquí (objeto de infra)
- 4.1.5.4
# En el Aggregate Root: def complete_order(self) -> None: self._status = OrderStatus.COMPLETED self._events.append(CompleteOrder(order_id=self.id)) # ← error: nombre en imperativo
- 4.1.5.5
# En el Command Handler: def handle(self, cmd: PlaceOrderCommand) -> None: order = self._repo.get(cmd.order_id) order.place() events = order.pull_events() order.notify_customers(events) # ← error: despacha desde el aggregate