Estructura del Proyecto
Arquitectura hexagonal con DDD: cómo se organiza el código y por qué.
Una estructura de proyecto bien diseñada no es un detalle estético: es la primera línea de defensa contra la entropía arquitectónica. Cuando los archivos están organizados por capa técnica (controllers/, models/, services/) en lugar de por dominio, el código crece acoplado: cualquier cambio en infraestructura atraviesa todo el sistema. La arquitectura hexagonal invierte esta lógica situando el dominio en el centro y convirtiendo todos los mecanismos externos —bases de datos, APIs, buses de eventos— en detalles intercambiables.
La estructura de directorios que se enseña en este módulo implementa tres ideas a la vez: Screaming Architecture (la estructura grita el dominio, no el framework), Arquitectura Hexagonal (los adaptadores implementan puertos definidos en el dominio/aplicación) y DDD táctico (cada Bounded Context es un paquete autónomo bajo `contexts/`). Estas tres ideas son complementarias y se refuerzan mutuamente.
El `shared_kernel/` es el único módulo que puede ser importado por múltiples Bounded Contexts. Debe mantenerse al mínimo absoluto: solo abstracciones base que todos necesitan (ValueObject, Entity, AggregateRoot, DomainEvent, Result). Si crece demasiado, se convierte en un monolito disfrazado. La disciplina aquí es cultural: agregar algo al shared_kernel requiere justificación explícita.
La regla de dependencia es la ley más importante de esta arquitectura y debe ser verificable automáticamente. `domain` no importa `application` ni `infrastructure`. `application` importa `domain` pero nunca `infrastructure` —solo sus puertos—. `infrastructure` importa ambos e implementa sus puertos. `main.py` es el único lugar que conoce las implementaciones concretas y arma el grafo de dependencias (Composition Root). Violar esta regla, aunque sea una vez, inicia una cadena de dependencias circulares difícil de revertir.
La separación entre comandos y queries en `application/` no es arbitraria: refleja CQRS a nivel de use case. Cada comando tiene su propio subdirectorio con `command.py` (el DTO de entrada) y `handler.py` (el use case). Cada query tiene además un `read_model.py` con el DTO de lectura optimizado. Esta separación hace que el código sea predecible: si buscas dónde se procesa una acción, sabes exactamente en qué directorio buscar.
Los tests también tienen una estructura deliberada: `unit/` para lógica de dominio pura sin I/O, `integration/` para adaptadores contra infraestructura real, y `e2e/` para flujos completos de API. Esta clasificación no es solo organización: determina la velocidad de ejecución, el tipo de fake o mock a usar, y qué se puede ejecutar en un pre-commit hook. La pirámide de tests es una consecuencia natural de esta estructura.
<project>/
├── pyproject.toml
├── uv.lock
├── .python-version
├── Makefile
├── README.md
├── src/<project>/
│ ├── __init__.py
│ ├── shared_kernel/
│ │ ├── domain/
│ │ │ ├── value_object.py
│ │ │ ├── entity.py
│ │ │ ├── aggregate_root.py
│ │ │ ├── domain_event.py
│ │ │ ├── result.py
│ │ │ └── errors.py
│ │ └── application/
│ │ ├── command.py
│ │ ├── query.py
│ │ ├── event_bus.py
│ │ └── unit_of_work.py
│ ├── contexts/
│ │ └── <bounded_context>/
│ │ ├── domain/
│ │ │ ├── model/
│ │ │ │ ├── <aggregate>.py
│ │ │ │ ├── <entity>.py
│ │ │ │ └── <value_object>.py
│ │ │ ├── events/
│ │ │ │ └── <something_happened>.py
│ │ │ ├── services/
│ │ │ │ └── <domain_service>.py
│ │ │ ├── repositories/
│ │ │ │ └── <aggregate>_repository.py
│ │ │ ├── factories/
│ │ │ │ └── <aggregate>_factory.py
│ │ │ └── specifications/
│ │ │ └── <rule>_specification.py
│ │ ├── application/
│ │ │ ├── commands/
│ │ │ │ └── <do_something>/
│ │ │ │ ├── command.py
│ │ │ │ └── handler.py
│ │ │ ├── queries/
│ │ │ │ └── <get_something>/
│ │ │ │ ├── query.py
│ │ │ │ ├── handler.py
│ │ │ │ └── read_model.py
│ │ │ └── ports/
│ │ │ ├── inbound/
│ │ │ │ └── <use_case>.py
│ │ │ └── outbound/
│ │ │ └── <gateway>.py
│ │ └── infrastructure/
│ │ ├── persistence/
│ │ │ ├── models.py
│ │ │ ├── mappers.py
│ │ │ └── sql_<aggregate>_repository.py
│ │ ├── api/
│ │ │ ├── controllers.py
│ │ │ └── schemas.py
│ │ ├── events/
│ │ │ └── <bus>_event_bus.py
│ │ └── acl/
│ │ └── <external>_acl.py
│ └── main.py
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
└── docker/
└── Dockerfile
# Dependency Rule (non-negotiable):
# infrastructure ──▶ application ──▶ domain
# infrastructure also implements ports defined in domain/application
# domain: NEVER imports application or infrastructure
# application: imports domain; NEVER imports infrastructure (ports only)
# main.py: only place that knows concrete implementations (Composition Root)Debugging lab
Detecta y corrige el error o la violación de diseño.
- 8.1.5.1
# BAD: inside contexts/<bounded_context>/domain/model/<aggregate>.py from sqlalchemy.orm import Session # <-- importing infrastructure in domain class <Aggregate>: def save(self, session: Session) -> None: session.add(self) session.commit()
- 8.1.5.2
# BAD: inside contexts/<bounded_context>/application/commands/<do_something>/handler.py from ..infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository class <DoSomething>Handler: def __init__(self) -> None: self.repo = Sql<Aggregate>Repository() # <-- concrete infra in application
- 8.1.5.3
# BAD: inside contexts/<bounded_context>/infrastructure/api/controllers.py # The controller contains business logic directly @router.post("/<aggregates>") async def create_<aggregate>(payload: CreatePayload, db: Session = Depends(get_db)): if payload.amount <= 0: raise HTTPException(400, "Amount must be positive") # domain rule in infra! aggregate = <Aggregate>(id=uuid4(), amount=payload.amount) db.add(aggregate) db.commit() return {"id": str(aggregate.id)}
- 8.1.5.4
# BAD: inside contexts/<bounded_context>/domain/repositories/<aggregate>_repository.py # Repository port leaks SQLAlchemy details from sqlalchemy.orm import Session class <Aggregate>Repository: def find_by_id(self, session: Session, id: <AggregateId>) -> <Aggregate>: ... def save(self, session: Session, aggregate: <Aggregate>) -> None: ...
- 8.1.5.5
# BAD: contexts/<bounded_context>/domain/model/<aggregate>.py # Domain imports from another bounded context directly from src.<project>.contexts.<other_context>.domain.model.<other_aggregate> import <OtherAggregate> class <Aggregate>: def __init__(self, other: <OtherAggregate>) -> None: self._other = other