Commands y Command Handlers
Un Command expresa una intención de cambio; el Handler la ejecuta orquestando el dominio.
Un Command es un objeto de datos inmutable que representa la intención de cambiar algo en el sistema: 'RegisterOrder', 'ApproveRequest', 'CancelSubscription'. No contiene comportamiento; es solo un mensaje con los datos necesarios para llevar a cabo la acción.
El Command Handler es el caso de uso para escrituras. Recibe un command, carga el aggregate correspondiente desde el repositorio, invoca el método de dominio apropiado, persiste el cambio y publica los eventos generados. No toma decisiones de negocio por sí mismo.
Los commands usan nombres en imperativo y en tiempo presente: 'RegisterOrder', no 'OrderRegistered' (eso sería un event). Esta distinción semántica ayuda a separar claramente las intenciones de escritura (commands) de los hechos ocurridos (events) y de las consultas (queries).
Un Command Handler tiene exactamente una responsabilidad. Si necesita coordinar con múltiples aggregates, la solución habitual es dividirlo en varios handlers coordinados por domain events, no agrandar el handler original. Mantener los handlers pequeños es clave para que sean testeables y comprensibles.
El retorno de un Command Handler debe ser mínimo. En la práctica devuelve None o el ID del aggregate creado. Devolver el estado completo del aggregate desde un handler de comando mezcla escritura y lectura, lo que contradice el principio CQRS.
# application/commands/<do_something>/
# Convención: una carpeta por caso de uso de escritura
# --- command.py --- (DTO de entrada)
from dataclasses import dataclass
@dataclass(frozen=True) # inmutable: el adapter HTTP lo crea, el handler lo lee
class <DoSomething>Command:
"""Intención de realizar <do_something> sobre <Aggregate>."""
aggregate_id: str # primitivos validados en la frontera (Pydantic schema)
# ... más campos según la intención
# NO: métodos de negocio aquí
# NO: importar dominio desde este archivo
# --- handler.py --- (Application Service / Use Case)
from <project>.contexts.<bc>.domain.repositories.<aggregate>_repository import <Aggregate>Repository
from <project>.contexts.<bc>.domain.model.<aggregate>_id import <Aggregate>Id
from <project>.shared_kernel.application.unit_of_work import UnitOfWork
from <project>.shared_kernel.application.event_bus import EventBus
from .command import <DoSomething>Command
class <DoSomething>Handler:
"""
Command Handler para <DoSomething>.
Orquesta repositorio, dominio y event bus.
NO contiene lógica de negocio.
"""
def __init__(
self,
repository: <Aggregate>Repository, # puerto (ABC), inyectado desde composition root
unit_of_work: UnitOfWork,
event_bus: EventBus,
) -> None:
self._repository = repository
self._uow = unit_of_work
self._event_bus = event_bus
def handle(self, command: <DoSomething>Command) -> None:
# 1. Reconstituir aggregate desde el repositorio (puerto, sin SQL aquí)
aggregate = self._repository.get_by_id(<Aggregate>Id(command.aggregate_id))
if aggregate is None:
raise <Aggregate>NotFoundError(command.aggregate_id)
# 2. Delegar la operación de dominio
aggregate.<do_something>() # el aggregate decide, protege invariantes
# 3. Persistir el aggregate como única unidad de trabajo
with self._uow:
self._repository.save(aggregate)
# 4. Publicar domain events registrados por el aggregate
for event in aggregate.pull_events():
self._event_bus.publish(event)
# --- shared_kernel/application/command.py --- (ABC base, opcional)
from abc import ABC, abstractmethod
from typing import Generic, TypeVar
C = TypeVar('C')
class CommandHandler(ABC, Generic[C]):
@abstractmethod
def handle(self, command: C) -> None: ...Debugging lab
Detecta y corrige el error o la violación de diseño.
- 5.2.5.1
@dataclass class RegisterOrderCommand: def validate(self): if not self.order_id: raise ValueError('missing id')
- 5.2.5.2
class RegisterOrderHandler: def handle(self, cmd): if cmd.total > 5000: raise BusinessRuleError('amount exceeds limit')
- 5.2.5.3
from sqlalchemy.orm import Session class CancelOrderHandler: def __init__(self, session: Session): self._session = session
- 5.2.5.4
def handle(self, cmd): order = self._repo.get_by_id(OrderId(cmd.order_id)) order.cancel() self._repo.save(order) return order # devolver el aggregate completo
- 5.2.5.5
def handle(self, cmd): order = self._order_repo.get_by_id(OrderId(cmd.order_id)) customer = self._customer_repo.get_by_id(CustomerId(cmd.customer_id)) order.assign_customer(customer) customer.add_order(order) self._order_repo.save(order) self._customer_repo.save(customer)