Clean Code y SOLID en DDD
Los cinco principios SOLID aplicados al diseño táctico de DDD con Python.
Los principios SOLID no son reglas abstractas de libro de texto: son diagnósticos de problemas concretos que aparecen en código real. En el contexto de DDD, SOLID y la arquitectura hexagonal se refuerzan mutuamente porque los dos responden a la misma preocupación fundamental: cómo mantener el código modificable a medida que el sistema crece sin que cada cambio rompa otras partes del sistema.
SRP (Single Responsibility Principle) en DDD se traduce en que cada Aggregate tiene exactamente una razón para cambiar: las reglas de negocio del concepto que representa. Un Aggregate que acumula responsabilidades de múltiples conceptos de negocio —por ejemplo, gestionar tanto el ciclo de vida de un `<Aggregate>` como las notificaciones de sus cambios— tiene múltiples razones para cambiar y se vuelve frágil. La señal de violación de SRP es cuando no puedes nombrar el Aggregate con un sustantivo singular del dominio.
OCP (Open/Closed Principle) en DDD se implementa mediante Domain Events y Specifications. Para extender el comportamiento del sistema ante una nueva regla de negocio, no modificas el Aggregate existente: publicas un Domain Event al que nuevos handlers se suscriben, o compones una nueva Specification. Si te encuentras abriendo constantemente un Aggregate para agregar `if/elif` que manejan nuevos casos, estás violando OCP.
LSP (Liskov Substitution Principle) es especialmente visible en los puertos. Si tienes un puerto `<Aggregate>Repository` y sus implementaciones —`Sql<Aggregate>Repository` y `InMemory<Aggregate>Repository`— el código que depende del puerto debe funcionar igual con cualquier implementación. Si el adaptador SQL lanza excepciones que el InMemory no lanza, o tiene precondiciones más estrictas, estás violando LSP y tus tests unitarios no serán representativos del comportamiento en producción.
ISP (Interface Segregation Principle) en DDD dice que los puertos deben ser pequeños y enfocados. Un puerto que combina `<Aggregate>Repository`, `UnitOfWork`, `EventBus` y `EmailGateway` en una sola interfaz obliga a todos los implementadores (incluyendo los fakes de test) a implementar métodos que no necesitan. Los puertos bien diseñados en DDD son estrechos: `<Aggregate>Repository` solo tiene `find_by_id` y `save`; `EventBus` solo tiene `publish`.
DIP (Dependency Inversion Principle) es la columna vertebral de la arquitectura hexagonal. El dominio y la aplicación dependen de abstracciones (puertos ABC/Protocol); la infraestructura proporciona las implementaciones concretas. La inversión es literal: en vez de que el dominio dependa de la base de datos, la base de datos depende del contrato que el dominio define. Esto hace que cambiar el motor de base de datos sea una cuestión de escribir un nuevo adaptador, sin tocar una línea de lógica de negocio.
# SRP — Aggregate with single responsibility
# BAD: <Aggregate> handles business logic AND sends notifications
class <Aggregate>: # has two reasons to change
def <perform_action>(self, <param>: <ValueObject>) -> None:
self._<attribute> = <param>
send_email(self._owner_email, '<action> done') # infra in domain!
# GOOD: Aggregate emits an event; a separate handler sends the email
class <Aggregate>:
def <perform_action>(self, <param>: <ValueObject>) -> None:
self._<attribute> = <param>
self._events.append(<SomethingHappened>(aggregate_id=self.id))
# OCP — extend via Specifications instead of modifying the Aggregate
from abc import ABC, abstractmethod
class <Rule>Specification(ABC):
@abstractmethod
def is_satisfied_by(self, aggregate: <Aggregate>) -> bool: ...
class <NewRule>Specification(<Rule>Specification):
def is_satisfied_by(self, aggregate: <Aggregate>) -> bool:
return aggregate.<attribute> <condition> # new rule without touching aggregate
# LSP — all Repository implementations honor the same contract
class <Aggregate>Repository(ABC):
@abstractmethod
def find_by_id(self, id: <AggregateId>) -> <Aggregate> | None: ...
@abstractmethod
def save(self, aggregate: <Aggregate>) -> None: ...
class InMemory<Aggregate>Repository(<Aggregate>Repository): # same contract
def find_by_id(self, id: <AggregateId>) -> <Aggregate> | None:
return self._store.get(id) # returns None if not found — same as SQL
def save(self, aggregate: <Aggregate>) -> None:
self._store[aggregate.id] = aggregate
# ISP — small focused ports
class <Aggregate>Repository(ABC): # only persistence concerns
@abstractmethod
def find_by_id(self, id: <AggregateId>) -> <Aggregate> | None: ...
@abstractmethod
def save(self, aggregate: <Aggregate>) -> None: ...
class EventBus(ABC): # only event publishing
@abstractmethod
def publish(self, events: list[DomainEvent]) -> None: ...
# DIP — handler depends on abstractions, not on concrete adapters
class <DoSomething>Handler:
def __init__(self, repo: <Aggregate>Repository, bus: EventBus) -> None:
self._repo = repo # ABC, not Sql<Aggregate>Repository
self._bus = bus # ABC, not <Bus>EventBus
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)
self._bus.publish(aggregate.pull_events())Debugging lab
Detecta y corrige el error o la violación de diseño.
- 8.3.5.1
# BAD: SRP violation — Aggregate handles business logic AND external calls # contexts/<bounded_context>/domain/model/<aggregate>.py import requests # infra in domain! class <Aggregate>: def <perform_action>(self, <param>: <ValueObject>) -> None: if not self._is_valid(<param>): raise <DomainError>("invalid <param>") self._<attribute> = <param> # Aggregate directly calls external service — second reason to change requests.post("https://external.service/notify", json={"id": str(self.id)})
- 8.3.5.2
# BAD: OCP violation — Aggregate modified every time a new rule is added # contexts/<bounded_context>/domain/model/<aggregate>.py class <Aggregate>: def can_<perform_action>(self, context: str) -> bool: if context == "normal": return self._<attribute> > 0 elif context == "premium": return self._<attribute> > 0 and self._tier == "premium" elif context == "vip": # new rule: every new context adds an elif return self._<attribute> > 0 and self._tier in ("premium", "vip") return False
- 8.3.5.3
# BAD: LSP violation — SQL adapter raises unexpected exception type # contexts/<bounded_context>/infrastructure/persistence/sql_<aggregate>_repository.py class Sql<Aggregate>Repository(<Aggregate>Repository): def find_by_id(self, id: <AggregateId>) -> <Aggregate>: result = self._session.query(<AggregateModel>).filter_by(id=str(id)).first() if result is None: raise ValueError(f"<Aggregate> {id} not found") # <-- breaks LSP! return <AggregateMapper>.to_domain(result) # The InMemory fake returns None (as the port contract says), # but the SQL adapter raises an exception — clients break in production. class InMemory<Aggregate>Repository(<Aggregate>Repository): def find_by_id(self, id: <AggregateId>) -> <Aggregate> | None: return self._store.get(id) # returns None — different behavior!
- 8.3.5.4
# BAD: ISP violation — fat port forces fakes to implement unused methods # shared_kernel/application/service_facade.py class ServiceFacade(ABC): # Repository methods @abstractmethod def find_<aggregate>_by_id(self, id: <AggregateId>) -> <Aggregate> | None: ... @abstractmethod def save_<aggregate>(self, aggregate: <Aggregate>) -> None: ... # Event bus methods @abstractmethod def publish_events(self, events: list) -> None: ... # Email gateway methods @abstractmethod def send_email(self, to: str, subject: str, body: str) -> None: ... # SMS gateway methods @abstractmethod def send_sms(self, to: str, message: str) -> None: ...
- 8.3.5.5
# BAD: DIP violation — handler depends on concrete infrastructure class # contexts/<bounded_context>/application/commands/<do_something>/handler.py # Direct import of concrete adapter — couples application to infrastructure from ...infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository from ...infrastructure.events.redis_event_bus import RedisEventBus class <DoSomething>Handler: def __init__(self) -> None: # Instantiates its own dependencies — impossible to test without real infra self._repo = Sql<Aggregate>Repository() self._bus = RedisEventBus()