Capas y Regla de Dependencia
El dominio no sabe que existe infraestructura; la infraestructura sí sabe del dominio.
La arquitectura por capas en DDD no es simplemente separar el código en carpetas: es una regla de dependencia estricta que define qué puede saber cada capa sobre las demás. La regla es simple pero absoluta: las dependencias de código siempre apuntan hacia el centro. El dominio es el centro y no importa nada de las capas externas. La infraestructura es la capa más externa y puede importar todo lo que necesite.
Las tres capas principales son dominio, aplicación e infraestructura. El dominio contiene el modelo del negocio: Aggregates, Value Objects, Domain Events, Domain Services, Repositories (como puertos/interfaces), Specifications. No tiene ninguna dependencia de framework. Si un archivo de la capa de dominio tiene un `import` de FastAPI, SQLAlchemy o cualquier librería de infraestructura, la regla de dependencia está rota.
La capa de aplicación orquesta el dominio: carga aggregates a través de puertos (interfaces de Repository), invoca métodos de dominio, persiste cambios, publica eventos. Los Application Services (implementados como Command Handlers y Query Handlers) viven aquí. Esta capa importa el dominio y usa sus puertos, pero NO importa implementaciones concretas de infraestructura —solo las interfaces abstractas.
La capa de infraestructura implementa los puertos definidos en el dominio y la aplicación. Los adapters concretos de bases de datos (SQLAlchemy), APIs (FastAPI), mensajería (Kafka, Redis, etc.) viven aquí. Esta capa sí puede importar frameworks. Implementa las interfaces del dominio/aplicación pero el dominio no sabe que estos adapters existen.
El punto de unión de todo es el Composition Root, generalmente `main.py` o un archivo de configuración de dependencias. Es el único lugar en todo el sistema donde se instancian las implementaciones concretas y se inyectan a través de las interfaces. Si en cualquier otro archivo del proyecto se crea una instancia de un adapter concreto de infraestructura, es una señal de que el wiring está mal hecho.
Esta arquitectura recibe distintos nombres: Hexagonal (Alistair Cockburn), Onion (Jeffrey Palermo), Clean (Robert C. Martin). Los detalles difieren pero el principio central es idéntico: el dominio en el centro, sin dependencias hacia afuera. En Python, la regla de dependencia se verifica con linters de arquitectura (como `import-linter` o `forbid-module-imports`) y se refuerza con tests que comprueban que ningún módulo del dominio tiene imports de infraestructura.
# The dependency rule in imports:
# domain → imports nothing outside itself
# application → imports domain only
# infrastructure → imports application + domain
# main.py → imports everything (composition root)
# domain/repositories/<aggregate>_repository.py
# PORT (abstract interface) — lives in domain, implemented in infra
from abc import ABC, abstractmethod
from ..<aggregate> import <Aggregate>, <AggregateId>
class <Aggregate>Repository(ABC): # Port — no infra import
@abstractmethod
def get(self, id: <AggregateId>) -> <Aggregate>: ...
@abstractmethod
def save(self, agg: <Aggregate>) -> None: ...
# application/commands/<do_something>/handler.py
# ORCHESTRATION — imports domain port, NOT the concrete adapter
from ...domain.repositories.<aggregate>_repository import <Aggregate>Repository
from ...domain.model.<aggregate> import <Aggregate>
class <DoSomethingHandler>:
def __init__(self, repo: <Aggregate>Repository) -> None:
self._repo = repo # injected: handler does not know the impl
def handle(self, cmd: <DoSomethingCommand>) -> None:
agg = self._repo.get(cmd.aggregate_id)
agg.<perform_action>(cmd.<param>)
self._repo.save(agg)
# infrastructure/persistence/sql_<aggregate>_repository.py
# ADAPTER — implements the port; imports SQLAlchemy here, NOT in domain
from sqlalchemy.orm import Session
from ...domain.repositories.<aggregate>_repository import <Aggregate>Repository
from .mappers import to_domain, to_orm
class Sql<Aggregate>Repository(<Aggregate>Repository):
def __init__(self, session: Session) -> None:
self._session = session
def get(self, id: <AggregateId>) -> <Aggregate>:
row = self._session.get(<AggregateORM>, str(id))
return to_domain(row)
def save(self, agg: <Aggregate>) -> None:
self._session.merge(to_orm(agg))
# main.py — COMPOSITION ROOT: only place that knows concrete impls
from infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository
from application.commands.<do_something>.handler import <DoSomethingHandler>
session = get_db_session()
repo = Sql<Aggregate>Repository(session)
handler = <DoSomethingHandler>(repo=repo) # DI via constructorDebugging lab
Detecta y corrige el error o la violación de diseño.
- 1.5.5.1
# In domain/model/<aggregate>.py from sqlalchemy.orm import Session class <Aggregate>: def save(self) -> None: session = Session() session.add(self) session.commit()
- 1.5.5.2
# In application/commands/<do_something>/handler.py from infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository class <Handler>: def handle(self, cmd: <Command>) -> None: repo = Sql<Aggregate>Repository() # concrete infra in application
- 1.5.5.3
# In main.py — wiring is scattered across multiple files # infrastructure/config.py: from application.commands.<do_something>.handler import <Handler> repo = Sql<Aggregate>Repository() handler = <Handler>(repo=repo) # wiring outside composition root
- 1.5.5.4
# Domain service importing an application port from application.ports.outbound.notification_gateway import NotificationGateway class <DomainService>: def __init__(self, gateway: NotificationGateway) -> None: self._gateway = gateway
- 1.5.5.5
# Unit test using real database from infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository def test_<aggregate>_<action>(): repo = Sql<Aggregate>Repository(real_db_session) agg = repo.get(<test_id>) agg.<perform_action>() assert agg._status == <expected_status>