Arquitectura orientada a eventos
Explora los Eventos de Dominio, el Event Sourcing y CQRS en DDD, cómo los eventos capturan ocurrencias de dominio, habilitan la integración asíncrona y respaldan los registros de auditoría
La Arquitectura orientada a eventos (EDA) es un patrón de arquitectura de software que usa eventos para disparar acciones y comunicar entre sí a servicios desacoplados. En DDD, los Eventos de Dominio se usan para modelar ocurrencias de negocio significativas, habilitando la consistencia eventual entre agregados y creando un registro de auditoría completo.
Un evento es "algo que sucedió en el pasado". En DDD, los Eventos de Dominio se nombran en tiempo pasado (p. ej., OrderPlaced, PaymentReceived, InventoryDepleted). Representan hechos que le importan al dominio.
Eventos de Dominio
Un Evento de Dominio es un registro de algo significativo que sucedió en el dominio. Captura el quién, el qué, el cuándo y (opcionalmente) el porqué de una ocurrencia de negocio.
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
@dataclass
class DomainEvent:
"""Base class for domain events."""
occurred_at: datetime = field(default_factory=datetime.now)
@dataclass
class OrderPlaced(DomainEvent):
order_id: str
customer_id: str
total_amount: float
item_count: int
@dataclass
class OrderConfirmed(DomainEvent):
order_id: str
confirmed_by: str
@dataclass
class PaymentReceived(DomainEvent):
order_id: str
transaction_id: str
amount: float
payment_method: str
@dataclass
class InventoryReserved(DomainEvent):
order_id: str
product_id: str
quantity: int
@dataclass
class OrderShipped(DomainEvent):
order_id: str
tracking_number: str
shipped_at: datetimeLanzamiento de Eventos de Dominio desde los Agregados
Los eventos de dominio se lanzan normalmente dentro de un agregado y se recopilan para publicarse después del commit de la transacción.
from dataclasses import dataclass, field
from enum import Enum
from typing import List
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
SHIPPED = "shipped"
DELIVERED = "delivered"
class Order:
def __init__(self, order_id: str, customer_id: str):
self._id = order_id
self._customer_id = customer_id
self._items: List["OrderLine"] = []
self._status = OrderStatus.PENDING
self._events: List[DomainEvent] = []
@property
def id(self) -> str:
return self._id
@property
def status(self) -> OrderStatus:
return self._status
def add_item(self, product_id: str, quantity: int, price: float) -> None:
if self._status != OrderStatus.PENDING:
raise ValueError("Cannot modify confirmed order")
self._items.append(OrderLine(product_id, quantity, price))
def confirm(self) -> None:
if self._status != OrderStatus.PENDING:
raise ValueError("Order is not pending")
if not self._items:
raise ValueError("Cannot confirm empty order")
self._status = OrderStatus.CONFIRMED
self._events.append(OrderConfirmed(
order_id=self._id,
confirmed_by="system"
))
def ship(self, tracking_number: str) -> None:
if self._status != OrderStatus.CONFIRMED:
raise ValueError("Can only ship confirmed orders")
self._status = OrderStatus.SHIPPED
self._events.append(OrderShipped(
order_id=self._id,
tracking_number=tracking_number,
shipped_at=datetime.now()
))
def pull_events(self) -> List[DomainEvent]:
events = list(self._events)
self._events.clear()
return eventsFlujo de eventos en un sistema DDD
Despacho y manejo de eventos
from typing import Callable, Type, Dict, List
class EventBus:
"""Simple synchronous event bus for in-process communication."""
def __init__(self):
self._handlers: Dict[Type, List[Callable]] = {}
def subscribe(self, event_type: Type, handler: Callable) -> None:
self._handlers.setdefault(event_type, []).append(handler)
def publish(self, event: DomainEvent) -> None:
for handler in self._handlers.get(type(event), []):
handler(event)
class OrderApplicationService:
def __init__(self, repo: "OrderRepository", bus: EventBus):
self._repo = repo
self._bus = bus
def confirm_order(self, order_id: str) -> None:
order = self._repo.find_by_id(order_id)
if not order:
raise ValueError(f"Order {order_id} not found")
order.confirm()
self._repo.save(order)
for event in order.pull_events():
self._bus.publish(event)
class InventoryEventHandler:
def __init__(self, repo: "InventoryRepository", bus: EventBus):
self._repo = repo
bus.subscribe(OrderConfirmed, self.on_order_confirmed)
def on_order_confirmed(self, event: OrderConfirmed) -> None:
print(f"[Inventory] Reserving stock for order {event.order_id}")
class NotificationEventHandler:
def __init__(self, bus: EventBus):
bus.subscribe(OrderConfirmed, self.on_order_confirmed)
bus.subscribe(OrderShipped, self.on_order_shipped)
def on_order_confirmed(self, event: OrderConfirmed) -> None:
print(f"[Notification] Sending confirmation for order {event.order_id}")
def on_order_shipped(self, event: OrderShipped) -> None:
print(f"[Notification] Sending tracking for order {event.order_id}")Event Sourcing (origen de eventos)
El Event Sourcing es un patrón de persistencia donde el estado se deriva de los eventos. En lugar de almacenar el estado actual de un agregado, almacenas la secuencia de eventos que llevaron a ese estado.
from dataclasses import dataclass
from typing import List, Protocol
class EventStore(Protocol):
def save_events(self, aggregate_id: str, events: List[DomainEvent],
expected_version: int) -> None: ...
def get_events(self, aggregate_id: str) -> List[DomainEvent]: ...
class EventSourcedOrder:
"""An aggregate rebuilt from its event stream."""
def __init__(self, order_id: str):
self._id = order_id
self._customer_id = ""
self._items: List[dict] = []
self._status = "pending"
self._version = 0
self._changes: List[DomainEvent] = []
@property
def id(self) -> str: return self._id
@property
def version(self) -> int: return self._version
@classmethod
def load_from_history(cls, events: List[DomainEvent]) -> "EventSourcedOrder":
order = cls.__new__(cls)
for event in events:
order._apply(event)
order._changes.clear()
return order
def place(self, customer_id: str, items: List[dict]) -> None:
event = OrderPlaced(
order_id=self._id,
customer_id=customer_id,
total_amount=sum(i["price"] * i["quantity"] for i in items),
item_count=len(items)
)
self._apply(event)
self._changes.append(event)
def confirm(self) -> None:
if self._status != "pending":
raise ValueError("Cannot confirm non-pending order")
event = OrderConfirmed(order_id=self._id, confirmed_by="system")
self._apply(event)
self._changes.append(event)
def _apply(self, event: DomainEvent) -> None:
if isinstance(event, OrderPlaced):
self._customer_id = event.customer_id
self._status = "pending"
elif isinstance(event, OrderConfirmed):
self._status = "confirmed"
elif isinstance(event, OrderShipped):
self._status = "shipped"
self._version += 1
def pull_changes(self) -> List[DomainEvent]:
changes = list(self._changes)
self._changes.clear()
return changes
class InMemoryEventStore:
def __init__(self):
self._streams: Dict[str, List[DomainEvent]] = {}
def save_events(self, aggregate_id: str, events: List[DomainEvent],
expected_version: int) -> None:
stream = self._streams.setdefault(aggregate_id, [])
if len(stream) != expected_version:
raise ValueError(f"Concurrency conflict for {aggregate_id}")
stream.extend(events)
def get_events(self, aggregate_id: str) -> List[DomainEvent]:
return list(self._streams.get(aggregate_id, []))Almacenamiento en Event Sourcing
CQRS: segregación de responsabilidades de comandos y consultas
CQRS separa los comandos (escrituras) de las consultas (lecturas). El modelo de escritura usa el modelo de dominio con agregados. El modelo de lectura usa vistas desnormalizadas optimizadas para la presentación.
# --- Command Side (Write Model) ---
@dataclass
class PlaceOrderCommand:
customer_id: str
items: List[dict]
class PlaceOrderHandler:
def __init__(self, repo: "OrderRepository", bus: EventBus, uow: "UnitOfWork"):
self._repo = repo
self._bus = bus
self._uow = uow
def handle(self, cmd: PlaceOrderCommand) -> str:
self._uow.begin()
try:
order_id = generate_order_id()
order = Order(order_id, cmd.customer_id)
for item in cmd.items:
order.add_item(item["product_id"], item["quantity"], item["price"])
order.confirm()
self._repo.save(order)
self._uow.commit()
for event in order.pull_events():
self._bus.publish(event)
return order_id
except Exception:
self._uow.rollback()
raise
# --- Query Side (Read Model) ---
@dataclass
class OrderSummary:
"""Denormalized read model optimized for display."""
order_id: str
customer_name: str
status: str
total: float
item_count: int
placed_at: str
class OrderQueryService:
"""Read-only service. No domain model, no business logic."""
def __init__(self, connection):
self._conn = connection
def get_order_summary(self, order_id: str) -> Optional[OrderSummary]:
cursor = self._conn.execute("""
SELECT o.order_id, c.name, o.status, o.total,
COUNT(ol.line_id) as item_count, o.placed_at
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
LEFT JOIN order_lines ol ON o.order_id = ol.order_id
WHERE o.order_id = ?
GROUP BY o.order_id
""", (order_id,))
row = cursor.fetchone()
if not row:
return None
return OrderSummary(**row)
def get_daily_sales_report(self, date: str) -> List[dict]:
cursor = self._conn.execute("""
SELECT o.status, COUNT(*) as count, SUM(o.total) as revenue
FROM orders o
WHERE DATE(o.placed_at) = ?
GROUP BY o.status
""", (date,))
return [dict(row) for row in cursor.fetchall()]CQRS añade complejidad. No añadas CQRS a menos que tengas una necesidad clara: formas de lectura y escritura distintas, requisitos de rendimiento o escala del equipo. Muchos sistemas funcionan perfectamente solo con repositorios.
Cuándo usar Event Sourcing
Factor | Event Sourcing adecuado | Event Sourcing excesivo |
|---|---|---|
Se requiere registro de auditoría | Sistemas financieros, cumplimiento | Herramientas internas |
Evolución compleja del estado | Procesos de negocio de larga duración | CRUD simple |
Depuración/valor | Necesidad de reproducir (replay) el estado histórico | El estado actual es suficiente |
Madurez del equipo | Experiencia con DDD | Nuevo en conceptos de DDD |
Requisitos de almacenamiento | Solo inserciones, inmutable | Actualizar el registro existente está bien |
# Projection: building a read model from events
class OrderSummaryProjection:
"""Builds denormalized read models from the event stream."""
def __init__(self, db_connection):
self._conn = db_connection
def apply(self, event: DomainEvent) -> None:
if isinstance(event, OrderPlaced):
self._conn.execute("""
INSERT INTO order_summaries
(order_id, status, total, item_count, placed_at)
VALUES (?, 'pending', ?, ?, ?)
""", (event.order_id, event.total_amount,
event.item_count, event.occurred_at))
elif isinstance(event, OrderConfirmed):
self._conn.execute("""
UPDATE order_summaries SET status = 'confirmed'
WHERE order_id = ?
""", (event.order_id,))
elif isinstance(event, OrderShipped):
self._conn.execute("""
UPDATE order_summaries SET status = 'shipped'
WHERE order_id = ?
""", (event.order_id,))
self._conn.commit()Sagas y gestores de proceso
Cuando un proceso de negocio abarca varios agregados (y posiblemente varios contextos), una Saga o un gestor de proceso (Process Manager) coordina el flujo.
@dataclass
class OrderProcessingSaga:
"""Coordinates the order fulfillment process across aggregates."""
def __init__(self, order_repo, payment_service,
inventory_service, shipping_service):
self._order_repo = order_repo
self._payment = payment_service
self._inventory = inventory_service
self._shipping = shipping_service
def handle_order_placed(self, event: OrderPlaced) -> None:
order = self._order_repo.find_by_id(event.order_id)
try:
self._payment.capture(order.total_amount)
self._inventory.reserve_items(event.order_id, order.items)
self._shipping.create_shipment(event.order_id)
order.mark_fulfilled()
except Exception as e:
order.mark_failed(str(e))
raise
finally:
self._order_repo.save(order)Convenciones de nombrado de eventos
Convención | Ejemplo | Cuándo usarla |
|---|---|---|
Verbo en tiempo pasado | OrderPlaced | DDD estándar |
Sustantivo + verbo | OrderConfirmed | DDD estándar |
Fuente + verbo | Sales.OrderConfirmed | Multicontexto |
Verbo + participio | PaymentReceived | Alternativa común |
Elige una convención de nombrado y manténla de forma consistente. El Lenguaje Ubicuo también se aplica a los eventos: usa las palabras que usan los expertos en dominio cuando describen lo que sucedió.
Versionado de eventos
Los eventos son contratos de datos. Evolucionan con el tiempo. Estrategias de versionado:
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
# V1: Original event
@dataclass
class OrderPlacedV1:
order_id: str
customer_id: str
total: float
# V2: Added discount field
@dataclass
class OrderPlacedV2:
order_id: str
customer_id: str
total: float
discount: float = 0.0
# Upcast function: converts V1 to V2
def upcast_order_placed(v1: OrderPlacedV1) -> OrderPlacedV2:
return OrderPlacedV2(
order_id=v1.order_id,
customer_id=v1.customer_id,
total=v1.total,
discount=0.0
)Beneficios de la Arquitectura orientada a eventos
Beneficio | Descripción |
|---|---|
Acoplamiento laxo | Los servicios se comunican mediante eventos, no con llamadas directas |
Registro de auditoría | Cada cambio de estado queda registrado |
Consultas temporales | Pueden responder "¿cuál era el estado en el momento X?" |
Capacidad de replay | Reconstruir el estado desde cero para depurar |
Procesamiento asíncrono | Sin bloqueos entre productor y consumidor |
Escalabilidad | Los manejadores de eventos pueden escalarse de forma independiente |
Pruebas de sistemas orientados a eventos
import pytest
from datetime import datetime
class TestEventDrivenOrder:
def test_order_raises_event_on_confirm(self):
order = Order("ORD-001", "CUST-001")
order.add_item("PROD-1", 2, 10.0)
order.confirm()
events = order.pull_events()
assert len(events) == 1
assert isinstance(events[0], OrderConfirmed)
assert events[0].order_id == "ORD-001"
def test_event_bus_dispatches_to_handlers(self):
bus = EventBus()
received = []
def handler(event):
received.append(event)
bus.subscribe(OrderConfirmed, handler)
event = OrderConfirmed(order_id="ORD-001", confirmed_by="test")
bus.publish(event)
assert len(received) == 1
assert received[0].order_id == "ORD-001"
def test_event_sourced_order_replay(self):
store = InMemoryEventStore()
order = EventSourcedOrder("ORD-001")
order.place("CUST-001", [{"product_id": "P1", "quantity": 2, "price": 10.0}])
store.save_events(order.id, order.pull_changes(), 0)
loaded = EventSourcedOrder.load_from_history(
store.get_events("ORD-001")
)
assert loaded.id == "ORD-001"
assert loaded.version == 1
def test_cqrs_read_model_updates(self):
bus = EventBus()
projection = FakeProjection()
bus.subscribe(OrderPlaced, projection.on_placed)
bus.subscribe(OrderConfirmed, projection.on_confirmed)
bus.publish(OrderPlaced(order_id="ORD-001", customer_id="CUST-001",
total_amount=100.0, item_count=2))
assert projection.statuses["ORD-001"] == "pending"
bus.publish(OrderConfirmed(order_id="ORD-001", confirmed_by="system"))
assert projection.statuses["ORD-001"] == "confirmed"
class FakeProjection:
def __init__(self):
self.statuses: dict = {}
def on_placed(self, event: OrderPlaced):
self.statuses[event.order_id] = "pending"
def on_confirmed(self, event: OrderConfirmed):
self.statuses[event.order_id] = "confirmed"Ejercicios de Práctica
-
Define eventos de dominio: para un sistema de gestión de una biblioteca, lista 8 eventos de dominio. Para cada evento, escribe el dataclass de Python con los campos apropiados.
-
Implementa la recolección de eventos: añade la recolección de eventos a un agregado
BankAccount. Incluye eventos paraAccountOpened,DepositMade,WithdrawalMadeyAccountFrozen. Muestra el métodopull_events. -
Implementación de un bus de eventos: implementa un bus de eventos asíncrono sencillo usando una queue.Queue en Python. Demuestra la suscripción de varios manejadores y la publicación de eventos.
-
Event Sourcing para BankAccount: implementa un agregado
BankAccountcon event sourcing. Incluyeload_from_history,applyy métodos de depósito y retiro. Prueba que reproducir (replay) los eventos produce el saldo correcto. -
Modelo de lectura de CQRS: diseña un modelo de lectura para un dashboard de reservas de hotel. ¿Qué consultas soportaría? ¿Qué tablas desnormalizadas crearías? Muestra el esquema SQL y el servicio de consultas.
-
Implementación de una proyección: implementa una
CustomerLoyaltyProjectionque escuche los eventosOrderPlacedy mantenga un total acumulado del gasto del cliente en una tabla separada. -
Diseño de una saga: diseña una saga para manejar el proceso de devoluciones en comercio electrónico: cuando se inicia una devolución, se debe procesar el reembolso, reponer el inventario y enviar una notificación. Muestra la saga como código de Python coordinando tres agregados.
-
Versionado de eventos: tienes un evento
InvoiceGeneratedV1 con los camposinvoice_id,customer_idytotal. En V2 necesitas añadirtax_amountydiscount_amount. Escribe el dataclass V2, una función de upcast y muestra cómo manejas ambas versiones en un manejador.
Has completado la Lección 7. La Arquitectura orientada a eventos, los Eventos de Dominio, el Event Sourcing y CQRS son patrones poderosos para construir sistemas escalables, auditables y desacoplados. Úsalos cuando la complejidad que resuelven supere la complejidad que introducen.