Use Cases (Application Services)
Los casos de uso orquestan el dominio sin contener lógica de negocio propia.
La capa de aplicación actúa como director de orquesta: recibe una intención del mundo exterior, coordina los objetos del dominio para cumplirla y devuelve un resultado. No sabe nada de HTTP, bases de datos ni colas de mensajes; solo conoce el dominio y sus puertos.
Un caso de uso representa una única intención del usuario o del sistema: 'registrar un pedido', 'aprobar una solicitud', 'cancelar una suscripción'. Ese nivel de granularidad hace que los casos de uso sean fáciles de leer, probar y razonar. Si un caso de uso hace dos cosas sin relación, es señal de que debería dividirse.
La regla de oro es que el caso de uso NO debe contener lógica de negocio. Las invariantes, las decisiones de dominio, los cálculos de negocio pertenecen al aggregate o al domain service. El caso de uso simplemente coordina: carga el aggregate, llama al método correspondiente, persiste y publica eventos.
Los casos de uso dependen de abstracciones (puertos) como el repositorio, el event bus y el unit of work. Nunca importan una clase de infraestructura directamente. Esta inversión de dependencias es lo que permite probar la capa de aplicación con fakes/in-memory sin levantar ninguna base de datos.
Cada caso de uso recibe su entrada como un Command o Query (objetos de datos simples, validados en la frontera), ejecuta la intención y devuelve un resultado mínimo. Para escrituras el retorno es vacío o un identificador; para lecturas devuelve un Read Model (DTO de lectura).
# use_cases/
# Cada caso de uso vive en su propio módulo dentro de application/
# --- Puerto (abstracción) — domain/repositories/<aggregate>_repository.py ---
from abc import ABC, abstractmethod
from <project>.contexts.<bc>.domain.model.<aggregate> import <Aggregate>
from <project>.contexts.<bc>.domain.model.<aggregate>_id import <Aggregate>Id
class <Aggregate>Repository(ABC): # Puerto: la aplicación solo conoce esta interfaz
@abstractmethod
def get_by_id(self, id: <Aggregate>Id) -> <Aggregate> | None: ...
@abstractmethod
def save(self, aggregate: <Aggregate>) -> None: ...
# --- Comando de entrada (DTO inmutable) ---
from dataclasses import dataclass
@dataclass(frozen=True)
class <DoSomething>Command:
aggregate_id: str # datos primitivos, validados en la frontera (adapter HTTP)
# ... otros campos
# --- Caso de uso (Application Service) ---
# application/commands/<do_something>/handler.py
from <project>.shared_kernel.application.unit_of_work import UnitOfWork
from <project>.shared_kernel.application.event_bus import EventBus
class <DoSomething>Handler:
def __init__(
self,
repository: <Aggregate>Repository, # puerto, no implementación concreta
unit_of_work: UnitOfWork, # puerto transaccional
event_bus: EventBus, # puerto de eventos
) -> None:
self._repository = repository
self._uow = unit_of_work
self._bus = event_bus
def handle(self, command: <DoSomething>Command) -> None:
# 1. Reconstituir el aggregate (o crear uno nuevo)
aggregate = self._repository.get_by_id(
<Aggregate>Id(command.aggregate_id)
)
if aggregate is None:
raise <Aggregate>NotFoundError(command.aggregate_id)
# 2. Delegar la decisión de negocio AL DOMINIO — nunca aquí
aggregate.<do_something>(...) # el aggregate protege sus invariantes
# 3. Persistir dentro de la unidad de trabajo
with self._uow:
self._repository.save(aggregate)
# 4. Publicar domain events (fuera de la transacción o dentro, según estrategia)
for event in aggregate.pull_events():
self._bus.publish(event)
# ANTI-PATRÓN — NO hacer esto en el handler:
# if aggregate.status == 'pending' and aggregate.amount > 1000: ← lógica de negocio
# aggregate.status = 'approved' ← mutación directa sin invarianteDebugging lab
Detecta y corrige el error o la violación de diseño.
- 5.1.5.1
def handle(self, cmd): self._repo.save(cmd.aggregate_id) # ¿qué viola esta línea?
- 5.1.5.2
def handle(self, cmd): if cmd.amount > 1000: aggregate.approve() else: aggregate.reject()
- 5.1.5.3
from sqlalchemy.orm import Session class <DoSomething>Handler: def __init__(self, session: Session): ...
- 5.1.5.4
def handle(self, cmd): agg_a = self._repo_a.get_by_id(cmd.id_a) agg_b = self._repo_b.get_by_id(cmd.id_b) agg_a.link(agg_b) agg_b.link(agg_a) self._repo_a.save(agg_a) self._repo_b.save(agg_b)
- 5.1.5.5
def handle(self, cmd): agg = self._repo.get_by_id(<Aggregate>Id(cmd.id)) agg.cancel() self._repo.save(agg) import smtplib smtplib.SMTP('smtp.example.com').sendmail(...) # notificar