Composition Root y Dependency Injection
El único lugar que conoce las implementaciones concretas.
El Composition Root es el punto de entrada del sistema donde se instancian todos los adapters concretos y se ensambla el grafo de dependencias. En una aplicación Python, ese lugar es `main.py` (o el equivalente de arranque de la aplicación). Su responsabilidad es única: conocer las implementaciones concretas de todos los puertos y pasárselas a los componentes que las necesitan. Fuera de `main.py`, ningún módulo de dominio ni de aplicación debe saber qué adapter concreto está usando.
La inyección de dependencias no debe ocurrir dentro del dominio ni de la aplicación. Si un Aggregate instancia su propio repositorio, o si un Application Service importa `Sql<Aggregate>Repository` directamente, la regla de dependencia está rota: la capa interna ha tomado control de qué adapter usa, y el Composition Root ha perdido su rol. El dominio y la aplicación deben recibir sus dependencias por constructor; nunca crearlas.
La inversión de dependencias (DIP) es el principio subyacente. Los módulos de alto nivel (dominio, aplicación) no dependen de módulos de bajo nivel (infraestructura); ambos dependen de abstracciones (puertos). Cuando el Application Service recibe un `<Aggregate>Repository` (ABC) en el constructor, no sabe ni le importa si la implementación es SQL, in-memory o un cliente HTTP. El Composition Root es el único lugar que sabe qué adapter concreto inyectar.
El grafo de dependencias se construye en orden inverso a la jerarquía de capas: primero se instancian los recursos de infraestructura (engine de base de datos, conexión al broker), luego los adapters que los usan (SqlRepository, BrokerEventBus), luego los handlers que reciben los adapters, y finalmente el router de FastAPI que registra los handlers como dependencias de las rutas. Este orden garantiza que cada componente recibe sus dependencias completamente construidas.
Frameworks de DI como `dependency-injector`, `punq` o el sistema de `Depends` de FastAPI automatizan el wiring pero no cambian el principio. Cuando el grafo es pequeño, el wiring manual en `main.py` es más legible y más fácil de seguir que un contenedor de DI. Cuando el grafo crece —muchos contextos, muchos adapters, configuración condicional— un contenedor aporta consistencia y reduce el código repetitivo. El criterio es la legibilidad y el mantenimiento, no la preferencia estética. Lo que nunca cambia es que el Composition Root es el único lugar que instancia adapters concretos.
# src/<project>/main.py — Composition Root
# ONLY place that imports concrete adapters.
# Domain and application layers are NOT imported here for logic — only for type hints.
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI
# 1. Infrastructure resources
from .contexts.<bounded_context>.infrastructure.persistence.models import Base
from .contexts.<bounded_context>.infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository
from .contexts.<bounded_context>.infrastructure.persistence.sql_unit_of_work import SqlUnitOfWork
from .contexts.<bounded_context>.infrastructure.events.<broker>_event_bus import <Broker>EventBus
from .contexts.<bounded_context>.infrastructure.api.router import router as <bounded_context>_router
# 2. Application handlers
from .contexts.<bounded_context>.application.commands.<do_something>.handler import <DoSomething>Handler
engine = create_engine('<DATABASE_URL>')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
# 3. Wire adapters into handlers
session = Session()
repo = Sql<Aggregate>Repository(session)
uow = SqlUnitOfWork(session)
event_bus = <Broker>EventBus(connection=<broker_connection>)
handler = <DoSomething>Handler(repo=repo, uow=uow, event_bus=event_bus)
# 4. Register in FastAPI
app = FastAPI()
app.include_router(<bounded_context>_router)
# BAD: do NOT do this in domain or application
# class <DoSomething>Handler: # application layer
# def __init__(self): # BAD: self-wiring
# self._repo = Sql<Aggregate>Repository() # concrete import in app layerDebugging lab
Detecta y corrige el error o la violación de diseño.
- 6.7.5.1
# domain/model/<aggregate>.py from infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository class <Aggregate>: def __init__(self) -> None: self._repo = Sql<Aggregate>Repository() # self-wiring in domain def <perform_action>(self, <param>: <ValueObject>) -> None: self._<field> = <param> self._repo.add(self)
- 6.7.5.2
# application/commands/<do_something>/handler.py from infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository from infrastructure.events.<broker>_event_bus import <Broker>EventBus class <DoSomething>Handler: def __init__(self) -> None: self._repo = Sql<Aggregate>Repository() # concrete import in app layer self._event_bus = <Broker>EventBus() # concrete import in app layer
- 6.7.5.3
# domain/model/<aggregate>.py from dependency_injector import containers, providers from infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository class <AggregateContainer>(containers.DeclarativeContainer): repo = providers.Factory(Sql<Aggregate>Repository) class <Aggregate>: container = <AggregateContainer>() _repo = container.repo()
- 6.7.5.4
# main.py — Composition Root from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.exc import SQLAlchemyError # infra import OK in main.py from .contexts.<bounded_context>.domain.repositories.<aggregate>_repository import <Aggregate>Repository from .contexts.<bounded_context>.infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository # Is it OK that main.py imports both the domain port AND the concrete adapter?
- 6.7.5.5
# contexts/<context_a>/main_a.py — Composition Root A repo_a = Sql<AggregateA>Repository(session_a) handler_a = <DoSomethingA>Handler(repo=repo_a) # contexts/<context_b>/main_b.py — Composition Root B repo_b = Sql<AggregateB>Repository(session_b) # different session! handler_b = <DoSomethingB>Handler(repo=repo_b) # Two separate entry points; handlers from A and B can never share a transaction