CQRS — Command Query Responsibility Segregation
Separar el modelo de escritura del de lectura para escalar y optimizar cada lado independientemente.
CQRS (Command Query Responsibility Segregation) es un principio que separa las operaciones que cambian estado (Commands) de las que lo leen (Queries). En su forma más básica, basta con tener handlers distintos para cada tipo; en su forma avanzada, los modelos de datos de lectura y escritura son completamente independientes.
El problema que CQRS resuelve es el del modelo unificado que intenta servir tanto para escrituras complejas (con invariantes y transacciones) como para lecturas optimizadas (con joins, proyecciones y filtros variados). Un aggregate bien diseñado es excelente para proteger reglas de negocio, pero terrible para generar listas paginadas con múltiples columnas calculadas.
En su forma simple, CQRS es casi transparente: Command Handlers usan el repositorio de escritura que reconstruye aggregates; Query Handlers usan un puerto de lectura que puede ir directo a SQL con una vista optimizada. No hay un store separado; solo rutas de código distintas para cada intención.
En su forma avanzada (módulo 7), el modelo de lectura vive en un almacén separado (read store), actualizado mediante proyecciones asíncronas disparadas por domain events. Esto habilita escalado independiente, pero introduce consistencia eventual. Esta versión solo tiene sentido cuando el volumen de lecturas o la complejidad de las proyecciones lo justifica.
El error más común es aplicar CQRS 'avanzado' por defecto. Para la mayoría de los sistemas, la versión simple (handlers separados, mismo store) aporta todos los beneficios de diseño sin la complejidad operacional de mantener dos almacenes sincronizados. Primero implementa la versión simple; escala si el problema real lo exige.
# CQRS simple — mismo store, handlers separados
# (la forma más común y recomendable como punto de partida)
# --- Lado de escritura (Command) ---
# application/commands/register_<aggregate>/handler.py
class Register<Aggregate>Handler:
def __init__(self, repo: <Aggregate>Repository, uow: UnitOfWork, bus: EventBus): ...
def handle(self, command: Register<Aggregate>Command) -> None:
aggregate = <Aggregate>.create(...) # dominio protege invariantes
with self._uow:
self._repo.save(aggregate)
for event in aggregate.pull_events():
self._bus.publish(event)
# --- Lado de lectura (Query) ---
# application/queries/get_<aggregates>/handler.py
class Get<Aggregates>Handler:
def __init__(self, read_port: <Aggregate>ReadPort): ...
def handle(self, query: Get<Aggregates>Query) -> list[<Aggregate>ReadModel]:
# Va directo a la fuente de lectura — sin reconstruir aggregates
return self._read_port.get_<aggregates>(query)
# --- Puerto de lectura (abstracción) ---
# application/ports/outbound/<aggregate>_read_port.py
from abc import ABC, abstractmethod
class <Aggregate>ReadPort(ABC):
@abstractmethod
def get_<aggregates>(self, query: Get<Aggregates>Query) -> list[<Aggregate>ReadModel]: ...
# --- Implementación simple (mismo store, consulta SQL optimizada) ---
# infrastructure/read_models/<aggregate>_sql_read_port.py
class <Aggregate>SqlReadPort(<Aggregate>ReadPort):
def __init__(self, session_factory): ...
def get_<aggregates>(self, query: Get<Aggregates>Query) -> list[<Aggregate>ReadModel]:
# Consulta SQL con JOINs y proyecciones sin reconstruir aggregates
# El dominio de escritura nunca toca esta clase
...
# NOTA: CQRS avanzado (store separado + proyecciones) → ver Módulo 7Debugging lab
Detecta y corrige el error o la violación de diseño.
- 5.4.5.1
class OrderService: def create_order(self, cmd): ... def get_order(self, id): return self._repo.get_by_id(id) # devuelve aggregate
- 5.4.5.2
class GetOrderHandler: def handle(self, query): order = self._repo.get_by_id(OrderId(query.id)) order.touch() # actualizar last_accessed self._repo.save(order) return OrderReadModel(id=str(order.id), ...)
- 5.4.5.3
class CreateOrderHandler: def handle(self, cmd) -> Order: order = Order.create(...) self._repo.save(order) return order # devolver aggregate para que la UI lo use
- 5.4.5.4
# CQRS 'avanzado' desde el primer día: # OrderWriteDB (PostgreSQL) + OrderReadDB (MongoDB) + event projector # ... para un sistema con 100 usuarios y un CRUD básico
- 5.4.5.5
class GetOrdersHandler: def handle(self, query): orders = self._repo.find_all() # carga todos los aggregates return [OrderReadModel(id=str(o.id), status=o.status.value) for o in orders]