Capítulo 6: Repositorios

5 de enero de 2025 Por: Artiko

dddrepositoriospythontactical-design

Capítulo 6: Repositorios

¿Qué es un Repositorio?

Un Repositorio proporciona una abstracción tipo colección para acceder a agregados, ocultando los detalles de persistencia. Actúa como puente entre el dominio y la infraestructura.

graph LR
    subgraph Dominio["DOMINIO"]
        AGG["Agregado\n(Pedido)"]
        REPO["«interface»\nRepositorio\n(Puerto)"]
    end

    subgraph Infra["INFRAESTRUCTURA"]
        IMPL["Repositorio\nSQLAlchemy\n(Adaptador)"]
        DB[("Base de Datos")]
    end

    AGG -->|"usa"| REPO
    IMPL -->|"implementa"| REPO
    IMPL -->|"persiste"| DB

“Un repositorio representa todos los objetos de cierto tipo como un conjunto conceptual. Actúa como una colección, excepto con capacidades de consulta más elaboradas.” — Eric Evans

Interfaz del Repositorio (Puerto)

from abc import ABC, abstractmethod
from typing import Optional
from uuid import UUID

class RepositorioPedidos(ABC):
    """
    Puerto: Define el contrato que debe cumplir cualquier implementación.
    Pertenece al dominio, no conoce detalles de persistencia.
    """

    @abstractmethod
    def guardar(self, pedido: "Pedido") -> None:
        """Persiste un agregado (insert o update)"""
        pass

    @abstractmethod
    def obtener(self, pedido_id: UUID) -> Optional["Pedido"]:
        """Recupera un agregado por su ID"""
        pass

    @abstractmethod
    def obtener_por_cliente(self, cliente_id: UUID) -> list["Pedido"]:
        """Recupera todos los pedidos de un cliente"""
        pass

    @abstractmethod
    def eliminar(self, pedido_id: UUID) -> None:
        """Elimina un agregado"""
        pass

    @abstractmethod
    def siguiente_id(self) -> UUID:
        """Genera un nuevo ID para un agregado"""
        pass

class RepositorioClientes(ABC):
    @abstractmethod
    def guardar(self, cliente: "Cliente") -> None:
        pass

    @abstractmethod
    def obtener(self, cliente_id: UUID) -> Optional["Cliente"]:
        pass

    @abstractmethod
    def buscar_por_email(self, email: "Email") -> Optional["Cliente"]:
        pass

    @abstractmethod
    def existe_email(self, email: "Email") -> bool:
        pass

Implementación en Memoria (Testing)

from uuid import uuid4

class RepositorioPedidosMemoria(RepositorioPedidos):
    """
    Implementación en memoria para testing.
    Simula comportamiento de base de datos.
    """

    def __init__(self):
        self._pedidos: dict[UUID, Pedido] = {}

    def guardar(self, pedido: Pedido) -> None:
        # Simular copia profunda como haría una BD
        self._pedidos[pedido.id] = pedido

    def obtener(self, pedido_id: UUID) -> Optional[Pedido]:
        return self._pedidos.get(pedido_id)

    def obtener_por_cliente(self, cliente_id: UUID) -> list[Pedido]:
        return [
            p for p in self._pedidos.values()
            if p.cliente_id == cliente_id
        ]

    def eliminar(self, pedido_id: UUID) -> None:
        self._pedidos.pop(pedido_id, None)

    def siguiente_id(self) -> UUID:
        return uuid4()

    # Métodos auxiliares para testing
    def limpiar(self) -> None:
        self._pedidos.clear()

    def contar(self) -> int:
        return len(self._pedidos)

    def todos(self) -> list[Pedido]:
        return list(self._pedidos.values())

Implementación con SQLAlchemy

from sqlalchemy import Column, String, Integer, ForeignKey, Numeric, Enum
from sqlalchemy.orm import Session, relationship, declarative_base
from uuid import UUID, uuid4
from decimal import Decimal

Base = declarative_base()

# === Modelos de Persistencia (Infraestructura) ===

class PedidoModelo(Base):
    __tablename__ = "pedidos"

    id = Column(String(36), primary_key=True)
    cliente_id = Column(String(36), nullable=False, index=True)
    estado = Column(String(20), nullable=False)
    direccion_calle = Column(String(200))
    direccion_ciudad = Column(String(100))
    direccion_cp = Column(String(10))
    lineas = relationship("LineaModelo", back_populates="pedido",
                         cascade="all, delete-orphan")

class LineaModelo(Base):
    __tablename__ = "lineas_pedido"

    id = Column(Integer, primary_key=True)
    pedido_id = Column(String(36), ForeignKey("pedidos.id"))
    producto_id = Column(String(36), nullable=False)
    nombre_producto = Column(String(200))
    cantidad = Column(Integer, nullable=False)
    precio_unitario = Column(Numeric(10, 2), nullable=False)
    pedido = relationship("PedidoModelo", back_populates="lineas")

# === Repositorio SQLAlchemy ===

class RepositorioPedidosSQLAlchemy(RepositorioPedidos):
    def __init__(self, session: Session):
        self._session = session

    def guardar(self, pedido: Pedido) -> None:
        modelo = self._a_modelo(pedido)
        self._session.merge(modelo)
        self._session.flush()

    def obtener(self, pedido_id: UUID) -> Optional[Pedido]:
        modelo = self._session.get(PedidoModelo, str(pedido_id))
        if modelo:
            return self._a_entidad(modelo)
        return None

    def obtener_por_cliente(self, cliente_id: UUID) -> list[Pedido]:
        modelos = (
            self._session.query(PedidoModelo)
            .filter(PedidoModelo.cliente_id == str(cliente_id))
            .all()
        )
        return [self._a_entidad(m) for m in modelos]

    def eliminar(self, pedido_id: UUID) -> None:
        modelo = self._session.get(PedidoModelo, str(pedido_id))
        if modelo:
            self._session.delete(modelo)

    def siguiente_id(self) -> UUID:
        return uuid4()

    # === Mapeo Modelo <-> Entidad ===

    def _a_modelo(self, pedido: Pedido) -> PedidoModelo:
        direccion = pedido.direccion_envio
        return PedidoModelo(
            id=str(pedido.id),
            cliente_id=str(pedido.cliente_id),
            estado=pedido.estado.value,
            direccion_calle=direccion.calle if direccion else None,
            direccion_ciudad=direccion.ciudad if direccion else None,
            direccion_cp=direccion.codigo_postal if direccion else None,
            lineas=[
                LineaModelo(
                    id=l.id,
                    producto_id=str(l.producto_id),
                    nombre_producto=l.nombre_producto,
                    cantidad=l.cantidad,
                    precio_unitario=l.precio_unitario,
                )
                for l in pedido.obtener_lineas()
            ],
        )

    def _a_entidad(self, modelo: PedidoModelo) -> Pedido:
        pedido = Pedido.__new__(Pedido)
        pedido.id = UUID(modelo.id)
        pedido.cliente_id = UUID(modelo.cliente_id)
        pedido._estado = EstadoPedido(modelo.estado)
        pedido._siguiente_linea_id = 1
        pedido._eventos = []

        # Reconstruir dirección
        if modelo.direccion_calle:
            pedido.direccion_envio = DireccionEnvio(
                calle=modelo.direccion_calle,
                ciudad=modelo.direccion_ciudad,
                codigo_postal=modelo.direccion_cp,
                pais="México"
            )
        else:
            pedido.direccion_envio = None

        # Reconstruir líneas
        pedido._lineas = []
        for lm in modelo.lineas:
            linea = LineaPedido(
                id=lm.id,
                producto_id=UUID(lm.producto_id),
                nombre_producto=lm.nombre_producto,
                cantidad=lm.cantidad,
                precio_unitario=Decimal(str(lm.precio_unitario)),
            )
            pedido._lineas.append(linea)
            pedido._siguiente_linea_id = max(
                pedido._siguiente_linea_id, lm.id + 1
            )

        return pedido

Patrón Unit of Work

El Unit of Work coordina la escritura de cambios y maneja transacciones.

from abc import ABC, abstractmethod
from contextlib import contextmanager

class UnitOfWork(ABC):
    """Coordina la persistencia de múltiples repositorios"""
    pedidos: RepositorioPedidos
    clientes: RepositorioClientes
    productos: RepositorioProductos

    @abstractmethod
    def __enter__(self) -> "UnitOfWork":
        pass

    @abstractmethod
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        pass

    @abstractmethod
    def commit(self) -> None:
        pass

    @abstractmethod
    def rollback(self) -> None:
        pass

    def recolectar_eventos(self) -> list:
        """Recolecta eventos de dominio de todos los agregados"""
        eventos = []
        # Los agregados guardan eventos pendientes
        return eventos

class UnitOfWorkSQLAlchemy(UnitOfWork):
    def __init__(self, session_factory):
        self._session_factory = session_factory
        self._session = None

    def __enter__(self) -> "UnitOfWork":
        self._session = self._session_factory()
        self.pedidos = RepositorioPedidosSQLAlchemy(self._session)
        self.clientes = RepositorioClientesSQLAlchemy(self._session)
        self.productos = RepositorioProductosSQLAlchemy(self._session)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type:
            self.rollback()
        self._session.close()

    def commit(self) -> None:
        self._session.commit()

    def rollback(self) -> None:
        self._session.rollback()

class UnitOfWorkMemoria(UnitOfWork):
    """Para testing - no requiere base de datos"""

    def __init__(self):
        self.pedidos = RepositorioPedidosMemoria()
        self.clientes = RepositorioClientesMemoria()
        self.productos = RepositorioProductosMemoria()
        self._committed = False

    def __enter__(self) -> "UnitOfWork":
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type:
            self.rollback()

    def commit(self) -> None:
        self._committed = True

    def rollback(self) -> None:
        self._committed = False

Especificaciones (Consultas Complejas)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Generic, TypeVar

T = TypeVar("T")

class Especificacion(ABC, Generic[T]):
    """Patrón Specification para consultas complejas"""

    @abstractmethod
    def es_satisfecha_por(self, candidato: T) -> bool:
        pass

    def y(self, otra: "Especificacion[T]") -> "Especificacion[T]":
        return EspecificacionY(self, otra)

    def o(self, otra: "Especificacion[T]") -> "Especificacion[T]":
        return EspecificacionO(self, otra)

    def no(self) -> "Especificacion[T]":
        return EspecificacionNo(self)

@dataclass
class EspecificacionY(Especificacion[T]):
    izq: Especificacion[T]
    der: Especificacion[T]

    def es_satisfecha_por(self, candidato: T) -> bool:
        return (
            self.izq.es_satisfecha_por(candidato)
            and self.der.es_satisfecha_por(candidato)
        )

# === Especificaciones de Pedido ===

@dataclass
class PedidoPorEstado(Especificacion["Pedido"]):
    estado: "EstadoPedido"

    def es_satisfecha_por(self, pedido: "Pedido") -> bool:
        return pedido.estado == self.estado

@dataclass
class PedidoPorCliente(Especificacion["Pedido"]):
    cliente_id: UUID

    def es_satisfecha_por(self, pedido: "Pedido") -> bool:
        return pedido.cliente_id == self.cliente_id

@dataclass
class PedidoConMontoMinimo(Especificacion["Pedido"]):
    monto_minimo: Decimal

    def es_satisfecha_por(self, pedido: "Pedido") -> bool:
        return pedido.total >= self.monto_minimo

@dataclass
class PedidoEnRangoFechas(Especificacion["Pedido"]):
    desde: date
    hasta: date

    def es_satisfecha_por(self, pedido: "Pedido") -> bool:
        fecha = pedido.fecha_creacion.date()
        return self.desde <= fecha <= self.hasta

# === Uso de Especificaciones ===

class RepositorioPedidosConEspecificacion(RepositorioPedidos):
    @abstractmethod
    def buscar(self, especificacion: Especificacion["Pedido"]) -> list["Pedido"]:
        pass

# Ejemplo de uso
def obtener_pedidos_pendientes_grandes(repo, cliente_id: UUID):
    spec = (
        PedidoPorCliente(cliente_id)
        .y(PedidoPorEstado(EstadoPedido.PENDIENTE))
        .y(PedidoConMontoMinimo(Decimal("1000")))
    )
    return repo.buscar(spec)

Uso en Casos de Uso

from dataclasses import dataclass
from uuid import UUID

@dataclass
class ConfirmarPedidoCommand:
    pedido_id: UUID

class ConfirmarPedidoHandler:
    def __init__(self, uow: UnitOfWork):
        self._uow = uow

    def ejecutar(self, comando: ConfirmarPedidoCommand) -> None:
        with self._uow:
            pedido = self._uow.pedidos.obtener(comando.pedido_id)

            if not pedido:
                raise PedidoNoEncontrado(comando.pedido_id)

            pedido.confirmar()

            self._uow.pedidos.guardar(pedido)
            self._uow.commit()

            # Publicar eventos después del commit
            for evento in pedido.obtener_eventos():
                self._publicar(evento)

@dataclass
class CrearPedidoCommand:
    cliente_id: UUID
    productos: list[dict]  # [{producto_id, cantidad}]

class CrearPedidoHandler:
    def __init__(self, uow: UnitOfWork):
        self._uow = uow

    def ejecutar(self, comando: CrearPedidoCommand) -> UUID:
        with self._uow:
            # Validar que el cliente existe
            cliente = self._uow.clientes.obtener(comando.cliente_id)
            if not cliente:
                raise ClienteNoEncontrado(comando.cliente_id)

            # Crear pedido con ID generado por repositorio
            pedido_id = self._uow.pedidos.siguiente_id()
            pedido = Pedido(id=pedido_id, cliente_id=comando.cliente_id)

            # Agregar productos
            for item in comando.productos:
                producto = self._uow.productos.obtener(item["producto_id"])
                if not producto:
                    raise ProductoNoEncontrado(item["producto_id"])

                pedido.agregar_producto(
                    producto_id=producto.id,
                    nombre=producto.nombre,
                    cantidad=item["cantidad"],
                    precio=producto.precio,
                )

            self._uow.pedidos.guardar(pedido)
            self._uow.commit()

            return pedido_id

Reglas de Repositorios

┌─────────────────────────────────────────────────────────────┐
│                  REGLAS DE REPOSITORIOS                      │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. UN REPOSITORIO POR AGGREGATE ROOT                       │
│     ✓ RepositorioPedidos                                    │
│     ✗ RepositorioLineasPedido (es entidad interna)          │
│                                                              │
│  2. INTERFAZ EN DOMINIO, IMPLEMENTACIÓN EN INFRAESTRUCTURA  │
│     dominio/repositorios.py  → interfaces abstractas        │
│     infra/sql_repositorios.py → implementaciones            │
│                                                              │
│  3. SIMULA UNA COLECCIÓN EN MEMORIA                         │
│     guardar(agregado)  → add/update                         │
│     obtener(id)        → get by id                          │
│     eliminar(id)       → remove                             │
│                                                              │
│  4. RECONSTRUYE AGREGADOS COMPLETOS                         │
│     ✓ Retorna Pedido con todas sus LineaPedido              │
│     ✗ Retorna solo datos parciales                          │
│                                                              │
│  5. NO EXPONE DETALLES DE PERSISTENCIA                      │
│     ✓ obtener(id: UUID)                                     │
│     ✗ obtener(query: "SELECT * FROM...")                    │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Antipatrones a Evitar

1. Repositorio Genérico

# MAL: Demasiado genérico, pierde significado de dominio
class RepositorioGenerico(ABC):
    def guardar(self, entidad: Any) -> None: pass
    def obtener(self, id: Any) -> Any: pass
    def buscar(self, criterio: dict) -> list: pass

# BIEN: Específico para cada agregado
class RepositorioPedidos(ABC):
    def guardar(self, pedido: Pedido) -> None: pass
    def obtener(self, pedido_id: UUID) -> Optional[Pedido]: pass
    def obtener_pendientes_de_envio(self) -> list[Pedido]: pass

2. Repositorio con Lógica de Negocio

# MAL: El repositorio tiene lógica de negocio
class RepositorioPedidosMal(RepositorioPedidos):
    def confirmar_pedido(self, pedido_id: UUID) -> None:
        pedido = self.obtener(pedido_id)
        pedido._estado = EstadoPedido.CONFIRMADO  # Lógica de negocio
        self.guardar(pedido)

# BIEN: Repositorio solo persiste
class RepositorioPedidosBien(RepositorioPedidos):
    def guardar(self, pedido: Pedido) -> None:
        # Solo persistencia, sin lógica
        modelo = self._a_modelo(pedido)
        self._session.merge(modelo)

3. Exponer Modelos de Persistencia

# MAL: Retorna modelo de SQLAlchemy
class RepositorioMal(RepositorioPedidos):
    def obtener(self, id: UUID) -> PedidoModelo:  # Modelo de BD
        return self._session.get(PedidoModelo, str(id))

# BIEN: Retorna entidad de dominio
class RepositorioBien(RepositorioPedidos):
    def obtener(self, id: UUID) -> Optional[Pedido]:  # Entidad
        modelo = self._session.get(PedidoModelo, str(id))
        return self._a_entidad(modelo) if modelo else None

Checklist de Repositorios

¿Hay un repositorio por cada aggregate root?
¿La interfaz está en el dominio?
¿La implementación está en infraestructura?
¿Se reconstruyen agregados completos?
¿Se ocultan detalles de persistencia?
¿Hay implementación en memoria para testing?
¿Se usa Unit of Work para transacciones?