Anti-Patrones en DDD
Los nueve errores más comunes en implementaciones DDD y cómo corregirlos.
Los anti-patrones en DDD son errores recurrentes que tienen nombre propio precisamente porque aparecen en casi todos los proyectos que intentan aplicar DDD sin una comprensión sólida de sus principios. Reconocerlos temprano es más valioso que corregirlos tarde: un Anti-Corruption Layer añadido después de seis meses de acoplamiento directo cuesta diez veces más que uno diseñado desde el principio.
El Anemic Domain Model es el anti-patrón más extendido y el más peligroso porque parece correcto: las clases se llaman `<Aggregate>` y `<ValueObject>`, pero son contenedores de datos con getters y setters públicos sin comportamiento real. La lógica de negocio migra silenciosamente a Services que terminan siendo procedimientos que manipulan estado externo, y el 'dominio' se convierte en un DTO glorificado. El indicador más claro es cuando un 'agregado' tiene más getters/setters que métodos de negocio.
El God Aggregate es el opuesto del Anemic Domain Model: un agregado que crece sin límites hasta absorber conceptos que deberían ser contextos independientes. Aparece cuando el equipo no delimita correctamente los Bounded Contexts y todo termina en un único modelo de dominio. La señal de alarma es cuando el agregado tiene más de cinco o seis colaboradores directos, o cuando sus métodos cubren más de dos o tres escenarios de uso radicalmente distintos.
La Fuga de Modelo (Model Leakage) ocurre cuando las entidades de dominio se exponen directamente en la API HTTP, ya sea como parámetros de entrada o como respuesta. Esto acopla el contrato HTTP al modelo interno de dominio: cualquier refactorización del dominio rompe los clientes de la API. La solución es usar Read Models (DTOs de lectura) en los Query Handlers y Pydantic schemas en los controllers de FastAPI.
El Dual-Write es uno de los problemas más sutiles y difíciles de detectar: ocurre cuando una operación escribe en la base de datos Y publica un evento en el mismo handler, pero sin una transacción atómica que garantice que ambas operaciones suceden o ninguna sucede. Si la escritura en DB falla después de publicar el evento, el sistema queda en un estado inconsistente. La solución correcta es el patrón Outbox: el evento se persiste en la misma transacción que el estado, y un proceso separado lo publica.
La sobre-ingeniería con CQRS y Event Sourcing es un anti-patrón de inversión de complejidad: aplicar patrones sofisticados en contextos donde no añaden valor. CQRS solo se justifica cuando las necesidades de lectura y escritura son tan distintas que un modelo unificado no puede servirlas eficientemente. Event Sourcing solo se justifica cuando el histórico de eventos ES el requerimiento de negocio, no cuando el equipo lo elige porque 'es la forma DDD de hacerlo'. El coste operacional de estos patrones es alto; aplícalos con evidencia de necesidad, no por aspiración.
# Anti-pattern 1: Anemic Domain Model — BEFORE
# domain/model/<aggregate>.py — only getters/setters, no behavior
class <Aggregate>:
def get_<attribute>(self): return self._<attribute>
def set_<attribute>(self, v): self._<attribute> = v # no validation!
# application/services/<aggregate>_service.py — logic lives outside domain
class <Aggregate>Service:
def <perform_action>(self, aggregate: <Aggregate>, param) -> None:
if param <= 0:
raise ValueError('invalid') # business rule outside the aggregate
aggregate.set_<attribute>(param)
# Anti-pattern 1: AFTER — rich domain model
class <Aggregate>:
def <perform_action>(self, <param>: <ValueObject>) -> None:
# Business rule lives inside the aggregate
if not <param>.is_valid():
raise <DomainError>('invalid <param>')
self._<attribute> = <param>
self._events.append(<SomethingHappened>(aggregate_id=self.id))
# Anti-pattern 2: Repository returns ORM model — BEFORE
class <Aggregate>Repository:
def find_by_id(self, id: str) -> <AggregateOrmModel>: # ORM object!
return self._session.query(<AggregateOrmModel>).filter_by(id=id).first()
# Anti-pattern 2: AFTER — repository maps to domain object
class Sql<Aggregate>Repository(<Aggregate>Repository):
def find_by_id(self, id: <AggregateId>) -> <Aggregate> | None:
orm_model = self._session.query(<AggregateOrmModel>).filter_by(id=str(id)).first()
if orm_model is None:
return None
return <AggregateMapper>.to_domain(orm_model) # always map to domain
# Anti-pattern 3: God Aggregate — BEFORE
class <GodAggregate>: # handles too many concepts
def __init__(self, profile, billing, subscription, notifications): ...
def update_profile(self): ...
def charge_payment(self): ...
def renew_subscription(self): ...
def send_notification(self): ...
# Anti-pattern 3: AFTER — separate Bounded Contexts
# contexts/identity/domain/model/<profile>.py
class <Profile>:
def update_<attribute>(self, value: <ValueObject>) -> None: ...
# contexts/billing/domain/model/<payment>.py
class <Payment>:
def charge(self, amount: <Amount>) -> None: ...Debugging lab
Detecta y corrige el error o la violación de diseño.
- 8.4.5.1
# BAD: Anemic Domain Model — logic lives outside the domain # contexts/<bounded_context>/domain/model/<aggregate>.py class <Aggregate>: def __init__(self, id, <attribute>): self.id = id self.<attribute> = <attribute> # public, no encapsulation # contexts/<bounded_context>/application/services/<aggregate>_service.py class <Aggregate>Service: def <perform_action>(self, aggregate: <Aggregate>, new_value) -> None: if new_value <= 0: # business rule outside domain! raise ValueError("value must be positive") if aggregate.<attribute> == new_value: raise ValueError("no change") aggregate.<attribute> = new_value # direct mutation from outside
- 8.4.5.2
# BAD: Repository returns ORM model — domain is coupled to infrastructure # contexts/<bounded_context>/infrastructure/persistence/sql_<aggregate>_repository.py from .models import <AggregateOrmModel> class Sql<Aggregate>Repository: def find_by_id(self, id: str) -> <AggregateOrmModel>: # returns ORM, not domain! return ( self._session.query(<AggregateOrmModel>) .filter_by(id=id) .first() ) # Callers now depend on SQLAlchemy internals: aggregate.column_name, lazy loading, etc.
- 8.4.5.3
# BAD: Repository contains business logic — violates separation of concerns # contexts/<bounded_context>/infrastructure/persistence/sql_<aggregate>_repository.py class Sql<Aggregate>Repository: def find_eligible_for_<action>(self) -> list[<Aggregate>]: # Business rule embedded in the repository query! return ( self._session.query(<AggregateOrmModel>) .filter( <AggregateOrmModel>.<attribute> > 0, <AggregateOrmModel>.status == "active", <AggregateOrmModel>.created_at < (datetime.now() - timedelta(days=30)), ) .all() )
- 8.4.5.4
# BAD: Model Leakage — domain entity exposed directly in API response # contexts/<bounded_context>/infrastructure/api/controllers.py from ...domain.model.<aggregate> import <Aggregate> @router.get("/<aggregates>/{id}") async def get_<aggregate>(id: str, repo: <Aggregate>Repository = Depends(get_repo)): aggregate = repo.find_by_id(<AggregateId>(id)) if aggregate is None: raise HTTPException(404) return aggregate # <-- exposes domain object as JSON! tight coupling
- 8.4.5.5
# BAD: Dual-write — DB write and event publish without atomicity # contexts/<bounded_context>/application/commands/<do_something>/handler.py class <DoSomething>Handler: def handle(self, command: <DoSomething>Command) -> None: aggregate = self._repo.find_by_id(command.<aggregate>_id) aggregate.<perform_action>(command.<param>) self._repo.save(aggregate) # step 1: write to DB events = aggregate.pull_events() self._bus.publish(events) # step 2: publish to broker # If step 2 crashes after step 1 commits: # - DB has the new state # - Event was never published # - Downstream consumers never notified — silent inconsistency