Repository Adapters
La implementación concreta del puerto Repository vive en infraestructura.
El patrón Repository tiene dos caras en la arquitectura hexagonal. El puerto abstracto vive en el dominio: es un `ABC` o `Protocol` que declara las operaciones de persistencia en términos del negocio (`get`, `add`, `list_by_<criteria>`). El adapter concreto vive en infraestructura: es la clase que implementa ese puerto usando SQLAlchemy, MongoDB, un cliente HTTP, o cualquier mecanismo de almacenamiento. El dominio y la aplicación dependen solo del puerto; el adapter concreto se inyecta desde el Composition Root.
El repositorio SQL gestiona tres responsabilidades concretas: abrir y cerrar el acceso a la sesión, traducir entre dominio y ORM mediante el mapper, y ejecutar las queries necesarias. La sesión SQLAlchemy se inyecta en el constructor; el repositorio no la crea. Las queries son SQL o SQLAlchemy ORM expressions y no contienen lógica de negocio: si necesitas 'todos los agregados en estado activo', la condición SQL `WHERE status = 'active'` es una query de infraestructura, pero si necesitas decidir cuáles de esos activos cumplen una invariante adicional, esa decisión es del dominio.
Lo que el repository adapter NO debe hacer es igualmente importante: no aplica especificaciones de dominio con lógica de negocio, no decide qué objetos son 'válidos' para ser persistidos, no lanza domain errors basados en reglas del negocio. El adapter es una bisagra técnica. Si el adapter necesita conocer las reglas del negocio para decidir qué persistir o qué retornar, la lógica está mal ubicada.
El repositorio in-memory es el adapter de test. Implementa exactamente el mismo puerto abstracto que el repositorio SQL, pero en lugar de usar SQLAlchemy usa un diccionario en memoria. Gracias a esto, los tests unitarios del Application Service y del Aggregate pueden correr sin base de datos, en microsegundos, sin setup de fixtures. El repositorio in-memory no es un mock de librería: es un adapter real que respeta el contrato del puerto.
El Unit of Work (UoW) gestiona la transacción que envuelve al repositorio. En lugar de que el repositorio haga `session.commit()` directamente, el UoW provee un contexto transaccional: `__enter__` abre la transacción, `__exit__` hace commit o rollback según si hubo excepción. El Application Service trabaja dentro de un UoW para garantizar que un conjunto de operaciones es atómico. El puerto del UoW vive en la capa de aplicación o shared_kernel; el adapter SQL vive en infraestructura.
# domain/repositories/<aggregate>_repository.py — Port (abstract)
from abc import ABC, abstractmethod
from uuid import UUID
from ..model.<aggregate> import <Aggregate>
class <Aggregate>Repository(ABC):
@abstractmethod
def get(self, aggregate_id: UUID) -> <Aggregate> | None: ...
@abstractmethod
def add(self, aggregate: <Aggregate>) -> None: ...
@abstractmethod
def list_all(self) -> list[<Aggregate>]: ...
# infrastructure/persistence/sql_<aggregate>_repository.py — SQL Adapter
from sqlalchemy.orm import Session
from ...domain.repositories.<aggregate>_repository import <Aggregate>Repository
from ...domain.model.<aggregate> import <Aggregate>
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: UUID) -> <Aggregate> | None:
row = self._session.get(<AggregateORM>, str(aggregate_id))
return to_domain(row) if row else None
def add(self, aggregate: <Aggregate>) -> None:
# BAD: do NOT add domain rules here
# if aggregate._status == 'invalid': raise DomainError(...) # violation
self._session.add(to_orm(aggregate))
def list_all(self) -> list[<Aggregate>]:
rows = self._session.query(<AggregateORM>).all()
return [to_domain(row) for row in rows]
# tests/unit/<bounded_context>/fakes/in_memory_<aggregate>_repository.py
from uuid import UUID
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: UUID) -> <Aggregate> | None:
return self._store.get(str(aggregate_id))
def add(self, aggregate: <Aggregate>) -> None:
self._store[str(aggregate.id)] = aggregate
def list_all(self) -> list[<Aggregate>]:
return list(self._store.values())Debugging lab
Detecta y corrige el error o la violación de diseño.
- 6.4.5.1
# infrastructure/persistence/sql_<aggregate>_repository.py class Sql<Aggregate>Repository(<Aggregate>Repository): def add(self, aggregate: <Aggregate>) -> None: if aggregate._<status> == '<invalid_status>': raise <DomainError>('Cannot add aggregate in invalid state') if not aggregate._<field>: raise <DomainError>('Field is required') self._session.add(to_orm(aggregate))
- 6.4.5.2
# infrastructure/persistence/sql_<aggregate>_repository.py class Sql<Aggregate>Repository(<Aggregate>Repository): def get(self, aggregate_id: UUID) -> <AggregateORM>: # wrong return type return self._session.get(<AggregateORM>, str(aggregate_id)) def list_all(self) -> list[<AggregateORM>]: # wrong return type return self._session.query(<AggregateORM>).all()
- 6.4.5.3
# application/commands/<do_something>/handler.py class <DoSomething>Handler: def handle(self, command: <DoSomething>Command) -> None: aggregates = self._repo.filter_by_status('<active_status>') # business filter in repo call for agg in aggregates: if agg._<field> == command.<field>: agg.<perform_action>(command.<param>)
- 6.4.5.4
# domain/model/<aggregate>.py from sqlalchemy.orm import Session class <Aggregate>: _session: Session # UoW inside domain! def save(self) -> None: uow = SqlUnitOfWork(self._session) with uow: uow.repo.add(self)
- 6.4.5.5
# Two repositories sharing a transaction without UoW async def handle(self, command: <DoSomething>Command) -> None: session = get_session() repo_a = Sql<AggregateA>Repository(session) repo_b = Sql<AggregateB>Repository(session) agg_a = repo_a.get(command.id_a) agg_b = repo_b.get(command.id_b) agg_a.<perform_action>() agg_b.<perform_action>() session.commit() # commit called directly in handler