Queries, Query Handlers y Read Models
Las consultas leen sin mutar estado; el Read Model devuelve solo lo que la UI necesita.
Una Query es la contraparte del Command: expresa una intención de lectura sin efectos secundarios. Preguntas como 'dame los pedidos pendientes del cliente X' o 'cuántos artículos quedan en stock' son queries. Al igual que los commands, son objetos de datos inmutables con los parámetros necesarios.
El Query Handler recibe la query y devuelve un Read Model: un DTO de lectura optimizado para la pantalla o el consumidor que hizo la pregunta. A diferencia del Command Handler, no carga un aggregate completo; puede ir directo a la fuente de datos más eficiente, incluso saltarse el dominio si la lectura es simple.
El Read Model no es el aggregate. Es una proyección plana y desnormalizada de los datos que el consumidor necesita, sin comportamiento ni invariantes. Puede tener campos calculados, combinar datos de múltiples fuentes o incluir solo un subconjunto de la información del aggregate.
En el contexto de CQRS, la separación entre query handlers y command handlers permite optimizar cada lado de forma independiente. El lado de lectura puede usar vistas SQL, caché, réplicas de base de datos o proyecciones desnormalizadas sin afectar la consistencia del modelo de escritura.
Los Query Handlers también dependen solo de puertos, pero con frecuencia usan un puerto de lectura diferente al repositorio de escritura. Este puerto de lectura puede ser tan simple como una interfaz que devuelve DTOs directamente desde SQL, sin reconstruir aggregates.
# application/queries/<get_something>/
# Convención: una carpeta por caso de uso de lectura
# --- read_model.py --- (DTO de salida)
from dataclasses import dataclass
from datetime import datetime
@dataclass(frozen=True) # inmutable: representa el snapshot de lo leído
class <GetSomething>ReadModel:
"""Proyección de lectura para <GetSomething>."""
id: str
# ... solo los campos que el consumidor necesita
# NO incluir objetos de dominio (Aggregate, Entity, ValueObject)
# NO incluir métodos de negocio
# --- query.py --- (DTO de consulta)
@dataclass(frozen=True)
class <GetSomething>Query:
"""Parámetros para la consulta <GetSomething>."""
# filtros / parámetros de paginación / identificadores
# ejemplo: aggregate_id: str | None = None
# --- Puerto de lectura (puede ser diferente al repositorio de escritura) ---
# domain/repositories/<aggregate>_read_repository.py (o en application/ports/)
from abc import ABC, abstractmethod
class <Aggregate>ReadPort(ABC):
@abstractmethod
def get_<something>(self, query: <GetSomething>Query) -> list[<GetSomething>ReadModel]: ...
# --- handler.py --- (Query Handler)
from .query import <GetSomething>Query
from .read_model import <GetSomething>ReadModel
class <GetSomething>Handler:
"""
Query Handler para <GetSomething>.
Lee sin efectos secundarios.
Devuelve un Read Model, nunca un Aggregate.
"""
def __init__(self, read_port: <Aggregate>ReadPort) -> None:
self._read_port = read_port # puerto de lectura (ABC)
def handle(self, query: <GetSomething>Query) -> list[<GetSomething>ReadModel]:
# NO cargar aggregates completos para leer — costoso e innecesario
return self._read_port.get_<something>(query)
# --- shared_kernel/application/query.py --- (ABC base, opcional)
from abc import ABC, abstractmethod
from typing import Generic, TypeVar
Q = TypeVar('Q')
R = TypeVar('R')
class QueryHandler(ABC, Generic[Q, R]):
@abstractmethod
def handle(self, query: Q) -> R: ...Debugging lab
Detecta y corrige el error o la violación de diseño.
- 5.3.5.1
class GetOrdersHandler: def handle(self, query): orders = self._repo.find_all() return [order for order in orders] # devuelve aggregates
- 5.3.5.2
class GetOrderDetailsHandler: def handle(self, query): order = self._repo.get_by_id(OrderId(query.order_id)) order.mark_as_viewed() # registrar que fue vista return OrderDetailsReadModel(...)
- 5.3.5.3
@dataclass class OrderReadModel: id: str aggregate: Order # referencia al aggregate completo
- 5.3.5.4
from sqlalchemy.orm import Session class GetOrdersHandler: def __init__(self, session: Session): self._session = session def handle(self, query): return self._session.query(OrderORM).all()
- 5.3.5.5
class GetAndUpdateOrderHandler: def handle(self, query): order = self._repo.get_by_id(OrderId(query.order_id)) order.refresh_cache() # 'por eficiencia' self._repo.save(order) return OrderReadModel(id=str(order.id), ...)