CQRS Avanzado
Modelos de lectura y escritura separados, proyecciones y stores independientes.
CQRS avanzado separa el modelo de comandos (lado de escritura, encargado de hacer cumplir invariantes del dominio) del modelo de consultas (lado de lectura, optimizado para la visualización). Los stores separados son opcionales pero comunes: el lado de escritura puede usar un event store append-only mientras el lado de lectura usa una base de datos relacional, documental o un índice de búsqueda. La separación física solo se justifica cuando el ratio lectura/escritura o los requisitos de escala lo demandan.
Las proyecciones son el puente entre ambos lados. Un handler de eventos se suscribe a los domain events emitidos por el aggregate y construye o actualiza vistas desnormalizadas en el read model. La proyección no llama al dominio ni toma decisiones de negocio: solo traduce eventos a representaciones optimizadas para el consumidor.
La consistencia eventual es la consecuencia natural de tener stores separados. El write side confirma la transacción y emite eventos; la proyección los procesa de forma asíncrona y actualiza el read model. El consumidor puede ver datos levemente desactualizados hasta que la proyección converge. Este tradeoff debe documentarse explícitamente en el contrato de la API.
Cuándo NO usar CQRS avanzado: aplicaciones CRUD simples sin lógica de dominio compleja, equipos pequeños sin experiencia en sistemas distribuidos, o cuando el ratio lectura/escritura no justifica dos modelos. Introducir stores separados, colas de eventos y proyecciones en una aplicación de 3 tablas es sobre-ingeniería que multiplica la complejidad operacional sin aportar valor real.
Los beneficios reales de CQRS avanzado aparecen a escala: el read side puede escalarse independientemente del write side, los modelos de lectura pueden moldearse exactamente para cada caso de uso sin comprometer el modelo de dominio, y la trazabilidad mediante eventos permite reconstruir cualquier estado pasado. Estos beneficios son reales pero tienen un coste; evalúalos antes de adoptarlo.
# application/commands/<do_something>_command.py — Command (write side)
# application/queries/<get_something>_query.py — Query (read side)
# application/handlers/<do_something>_handler.py — Command Handler
# application/projections/<readmodel>_projection.py — Projection (Event → Read Model)
@dataclass(frozen=True)
class <DoSomething>Command:
aggregate_id: <AggregateId>
<param>: <ValueObject>
@dataclass(frozen=True)
class <GetSomething>Query:
aggregate_id: <AggregateId>
# Command handler — write side
class <DoSomething>Handler:
def __init__(self, repo: <AggregateRepository>) -> None:
self._repo = repo
def handle(self, cmd: <DoSomething>Command) -> None:
aggregate = self._repo.find(cmd.aggregate_id)
aggregate.<perform_action>(cmd.<param>)
self._repo.save(aggregate)
# Query handler — read side
class <GetSomething>Handler:
def __init__(self, read_store: <ReadModelStore>) -> None:
self._store = read_store
def handle(self, query: <GetSomething>Query) -> <ReadModelDto>:
return self._store.find(query.aggregate_id)
# Projection — subscribes to events, updates read model
class <ReadModel>Projection:
def __init__(self, store: <ReadModelStore>) -> None:
self._store = store
def on_<something_happened>(self, event: <SomethingHappened>) -> None:
# Upsert denormalized read model from event data
self._store.upsert(<ReadModelDto>(id=event.aggregate_id, ...))Debugging lab
Detecta y corrige el error o la violación de diseño.
- 7.1.5.1
# Bug: query handler reads from the aggregate repository and returns the aggregate directly class GetSomethingHandler: def __init__(self, repo: AggregateRepository) -> None: self._repo = repo def handle(self, query: GetSomethingQuery) -> Aggregate: return self._repo.find(query.aggregate_id)
- 7.1.5.2
# Bug: command handler also updates the read model table inside the same transaction class DoSomethingHandler: def handle(self, cmd: DoSomethingCommand) -> None: aggregate = self._repo.find(cmd.aggregate_id) aggregate.perform_action(cmd.param) self._repo.save(aggregate) # Also updating read model directly in same transaction self._read_store.upsert(ReadModelDto(id=cmd.aggregate_id, value=cmd.param))
- 7.1.5.3
# Bug: projection raises a DomainError when event data is considered "invalid" class ReadModelProjection: def on_something_happened(self, event: SomethingHappened) -> None: if not event.value: raise DomainError("Invalid event: value cannot be empty") self._store.upsert(ReadModelDto(id=event.aggregate_id, value=event.value))
- 7.1.5.4
# Bug: read model DTO exposing the raw internal aggregate dictionary class GetSomethingHandler: def handle(self, query: GetSomethingQuery) -> dict: aggregate = self._repo.find(query.aggregate_id) return aggregate.__dict__ # Exposing raw internal state
- 7.1.5.5
# Bug: separate CQRS with two DB stores added from day one to a simple 3-table CRUD app # Setting up: PostgreSQL write store + MongoDB read store + event queue + projections # for a basic admin panel with no complex domain logic