Puertos y Adapters (Arquitectura Hexagonal)
El dominio en el centro; la infraestructura en el borde.
La Arquitectura Hexagonal, propuesta por Alistair Cockburn, organiza el software en tres zonas concéntricas: el dominio en el centro, la capa de aplicación como orquestadora, y la infraestructura en el borde. El patrón de Puertos y Adapters formaliza cómo estas zonas se comunican: un puerto es una interfaz abstracta que el dominio o la aplicación declara, y un adapter es la implementación concreta que vive en infraestructura. La clave es que el dominio nunca sabe qué adapter lo sirve.
Existen dos familias de puertos según la dirección del flujo. Los puertos driving (o primarios) son los que el mundo exterior usa para activar la aplicación: una API HTTP, un comando CLI, un mensaje de cola. Los puertos driven (o secundarios) son los que la aplicación usa para interactuar con el exterior: un repositorio de datos, un bus de eventos, un cliente de servicio externo. Un driving adapter convierte una petición externa en una llamada a un caso de uso; un driven adapter implementa el contrato que la aplicación espera.
La regla de dependencia es el principio más importante de la arquitectura hexagonal: las dependencias siempre apuntan hacia adentro. El adapter de infraestructura conoce el puerto de dominio y lo implementa. El dominio no sabe nada del adapter. Esto es inversión de dependencias aplicada a escala arquitectónica: la capa más estable (el dominio) no depende de la más volátil (la infraestructura). Si el adapter cambia —de PostgreSQL a MongoDB, de RabbitMQ a Redis— el dominio permanece intacto.
Las ventajas más tangibles son la testabilidad y la sustituibilidad. Como el dominio solo depende de abstracciones, se puede reemplazar cualquier adapter por un doble de test (un repositorio in-memory, un event bus que solo colecciona eventos) sin tocar ni una línea del dominio. La sustituibilidad permite migrar de proveedor de infraestructura sin regresiones en la lógica de negocio, ya que la única pieza que cambia es el adapter concreto.
En Python idiomático, los puertos se expresan con `ABC` (Abstract Base Class) o con `Protocol` de `typing`. `ABC` fuerza herencia explícita y falla en tiempo de instanciación si el adapter no implementa todos los métodos; `Protocol` permite structural subtyping y es útil cuando el adapter es una clase de terceros que no puede heredar de tu ABC. Para puertos internos del proyecto, `ABC` comunica la intención más claramente; para adaptar código externo sin modificarlo, `Protocol` es la herramienta correcta.
# domain/repositories/<aggregate>_repository.py — Port (driven, abstract)
# Define the contract. NEVER import SQLAlchemy, Redis, or any infra here.
from abc import ABC, abstractmethod
from ..<model> import <Aggregate>
from ..<value_object> import <AggregateId>
class <Aggregate>Repository(ABC):
@abstractmethod
def get(self, aggregate_id: <AggregateId>) -> <Aggregate> | None: ...
@abstractmethod
def add(self, aggregate: <Aggregate>) -> None: ...
# BAD: do NOT do this in the domain port
# from sqlalchemy.orm import Session # <-- infra import in domain = violation
# infrastructure/persistence/sql_<aggregate>_repository.py — Adapter (driven, concrete)
# Depends on domain port. May import SQLAlchemy freely.
from sqlalchemy.orm import Session
from ...domain.repositories.<aggregate>_repository import <Aggregate>Repository
from .mappers import to_domain, to_orm
from .models import <AggregateORM>
class Sql<Aggregate>Repository(<Aggregate>Repository):
def __init__(self, session: Session) -> None:
self._session = session
def get(self, aggregate_id: <AggregateId>) -> <Aggregate> | None:
row = self._session.get(<AggregateORM>, str(aggregate_id))
return to_domain(row) if row else None
def add(self, aggregate: <Aggregate>) -> None:
self._session.add(to_orm(aggregate))
# tests/fakes/in_memory_<aggregate>_repository.py — Mock Adapter (for unit tests)
from collections import defaultdict
from ...domain.repositories.<aggregate>_repository import <Aggregate>Repository
class InMemory<Aggregate>Repository(<Aggregate>Repository):
def __init__(self) -> None:
self._store: dict[str, <Aggregate>] = {}
def get(self, aggregate_id: <AggregateId>) -> <Aggregate> | None:
return self._store.get(str(aggregate_id))
def add(self, aggregate: <Aggregate>) -> None:
self._store[str(aggregate.id)] = aggregateDebugging lab
Detecta y corrige el error o la violación de diseño.
- 6.1.5.1
# domain/model/<aggregate>.py from sqlalchemy.orm import Session class <Aggregate>: def save(self, session: Session) -> None: session.add(self) session.commit()
- 6.1.5.2
# domain/repositories/<aggregate>_repository.py from sqlalchemy.orm import Session from ..infrastructure.persistence.models import <AggregateORM> class <Aggregate>Repository(ABC): @abstractmethod def get(self, id: str) -> <AggregateORM>: ...
- 6.1.5.3
# infrastructure/persistence/sql_<aggregate>_repository.py class Sql<Aggregate>Repository(<Aggregate>Repository): def add(self, aggregate: <Aggregate>) -> None: if aggregate._status == 'inactive': raise <DomainError>('Cannot persist inactive aggregate') self._session.add(to_orm(aggregate))
- 6.1.5.4
# 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 adapter instantiated here
- 6.1.5.5
# Two adapters for the same port — one leaks infra detail in return type from .models import <AggregateORM> class Sql<Aggregate>Repository(<Aggregate>Repository): def list_all(self) -> list[<AggregateORM>]: # returns ORM type return self._session.query(<AggregateORM>).all() class InMemory<Aggregate>Repository(<Aggregate>Repository): def list_all(self) -> list[<Aggregate>]: # returns domain type return list(self._store.values())