Repositories
Puerto de acceso a aggregates: una abstracción de colección en el dominio, sin implementación.
El Repository es un puerto: una interfaz abstracta definida en la capa de dominio que simula una colección en memoria de aggregates. La capa de dominio declara qué necesita (`get_by_id`, `save`, `find_by_<criteria>`); la capa de infraestructura lo implementa. Esta separación es la esencia del principio de inversión de dependencias.
Existe exactamente un Repository por Aggregate Root. No se crean repositorios para entidades hijas: siempre se accede a ellas a través de la raíz. Esta regla refuerza el límite de consistencia del aggregate y evita modificar entidades huérfanas fuera de su contexto transaccional.
El Repository habla el lenguaje del dominio. Sus métodos retornan aggregates o colecciones de aggregates, nunca modelos ORM, filas de base de datos ni DTOs. El mapeo entre el modelo de dominio y el modelo de persistencia ocurre en el adaptador de infraestructura, invisible para el dominio.
La interfaz abstracta del Repository pertenece al dominio (`domain/repositories/`). La implementación concreta pertenece a la infraestructura (`infrastructure/persistence/`). Esta separación permite testear el dominio y la capa de aplicación con fakes in-memory rápidos, sin levantar una base de datos.
En Python, el puerto se implementa con `ABC` (Abstract Base Class) o con `Protocol` (duck typing estructural). Ambas opciones son válidas; `Protocol` es más flexible para testing porque no requiere herencia explícita en los fakes. La elección debe ser consistente en todo el proyecto.
El Repository no es un DAO (Data Access Object) genérico con métodos `create/read/update/delete`. Es una abstracción orientada al negocio: sus métodos tienen nombres significativos que reflejan intenciones del dominio, y solo expone las operaciones que el dominio realmente necesita.
# domain/repositories/<aggregate>_repository.py
# PUERTO: interfaz abstracta. Solo declaraciones, cero lógica de infra.
# NO importar SQLAlchemy, bases de datos ni frameworks.
from abc import ABC, abstractmethod
from uuid import UUID
from typing import Optional
from ..<model>.<aggregate> import <Aggregate>
class <Aggregate>Repository(ABC): # Puerto (abstracción de dominio)
"""
Colección abstracta de <Aggregate>.
La implementación concreta vive en infrastructure/persistence/.
"""
@abstractmethod
def get_by_id(self, aggregate_id: UUID) -> Optional[<Aggregate>]:
"""Devuelve el aggregate o None. NUNCA devuelve un modelo ORM."""
...
@abstractmethod
def save(self, aggregate: <Aggregate>) -> None:
"""Persiste (insert o update). El aggregate ya fue validado."""
...
@abstractmethod
def find_by_<criteria>(self, <param>: <Type>) -> list[<Aggregate>]:
"""Consulta específica de dominio; nombre en lenguaje ubicuo."""
...
# NO agregar: bulk_create, execute_raw_sql, get_session, flush
# ── Alternativa con Protocol (duck typing estructural) ──────────────────────
from typing import Protocol, runtime_checkable
@runtime_checkable
class <Aggregate>Repository(Protocol): # Puerto con Protocol
def get_by_id(self, aggregate_id: UUID) -> Optional[<Aggregate>]: ...
def save(self, aggregate: <Aggregate>) -> None: ...
def find_by_<criteria>(self, <param>: <Type>) -> list[<Aggregate>]: ...
# ── Fake in-memory (para tests) ─────────────────────────────────────────────
# tests/fakes/<aggregate>_repository_fake.py
class <Aggregate>RepositoryFake(<Aggregate>Repository):
def __init__(self) -> None:
self._store: dict[UUID, <Aggregate>] = {}
def get_by_id(self, aggregate_id: UUID) -> Optional[<Aggregate>]:
return self._store.get(aggregate_id)
def save(self, aggregate: <Aggregate>) -> None:
self._store[aggregate.id] = aggregate
def find_by_<criteria>(self, <param>: <Type>) -> list[<Aggregate>]:
return [a for a in self._store.values() if a.<criteria> == <param>]Debugging lab
Detecta y corrige el error o la violación de diseño.
- 4.3.5.1
# domain/repositories/order_repository.py from sqlalchemy.orm import Session # ← error: import de infra en el puerto class OrderRepository(ABC): def __init__(self, session: Session) -> None: # ← error: sesión en el puerto self._session = session @abstractmethod def get_by_id(self, order_id: UUID) -> Optional[Order]: ...
- 4.3.5.2
# infrastructure/persistence/sql_order_repository.py class SqlOrderRepository(OrderRepository): def get_by_id(self, order_id: UUID) -> Optional[OrderOrmModel]: # ← retorna ORM model return self._session.query(OrderOrmModel).filter_by(id=order_id).first()
- 4.3.5.3
# domain/repositories/product_repository.py class ProductRepository(ABC): @abstractmethod def get_by_id(self, product_id: UUID) -> Optional[Product]: ... @abstractmethod def get_variant_by_id(self, variant_id: UUID) -> Optional[ProductVariant>]: # ← error: entidad hija ...
- 4.3.5.4
# domain/repositories/invoice_repository.py class InvoiceRepository(ABC): @abstractmethod def get_by_id(self, invoice_id: UUID) -> Optional[Invoice]: ... @abstractmethod def execute_raw_query(self, sql: str) -> list[dict]: # ← error en el puerto ...
- 4.3.5.5
# domain/repositories/shipment_repository.py class ShipmentRepository(ABC): @abstractmethod def save(self, shipment: Shipment) -> None: ... @abstractmethod def begin_transaction(self) -> None: # ← error: gestión transaccional en el puerto ... @abstractmethod def commit(self) -> None: ...