IF0100 - Programación OO II

4. Patrón Repository - Abstracción de Persistencia

4.1 Interfaz Repository Genérica

# domain/repositories/base.py
from abc import ABC, abstractmethod  # ABC = Abstract Base Class
from typing import TypeVar, Generic, List, Optional

T = TypeVar('T')  # Tipo genérico para cualquier entidad

class Repository(ABC, Generic[T]):
    """
    Interfaz base para todos los repositorios.
    
    Define las operaciones CRUD básicas que cualquier
    repositorio debe implementar.
    
    T es el tipo de entidad (Usuario, Tarea, etc.)
    """

    @abstractmethod
    def guardar(self, entidad: T) -> T:
        """
        Guarda una entidad.
        Si la entidad no tiene ID, la asigna.
        Si ya tiene ID, actualiza.
        
        Returns:
            La entidad guardada (con ID si era nueva)
        """
        pass

    @abstractmethod
    def obtener_por_id(self, id: int) -> Optional[T]:
        """
        Busca una entidad por su ID.
        
        Returns:
            La entidad si existe, None si no
        """
        pass

    @abstractmethod
    def obtener_todos(self) -> List[T]:
        """
        Obtiene todas las entidades.
        
        Returns:
            Lista de entidades (vacía si no hay)
        """
        pass

    @abstractmethod
    def eliminar(self, id: int) -> bool:
        """
        Elimina una entidad por ID.
        
        Returns:
            True si se eliminó, False si no existía
        """
        pass

    @abstractmethod
    def existe(self, id: int) -> bool:
        """
        Verifica si existe una entidad con ese ID.
        
        Returns:
            True si existe, False si no
        """
        pass

4.2 Implementación en Memoria (para Tests)

# infrastructure/repositories/usuario_repo.py
from typing import Dict, List, Optional
from domain.repositories.base import Repository
from domain.models.usuario import Usuario

class UsuarioMemoryRepository(Repository[Usuario]):
    """
    Implementación en memoria del repositorio de usuarios.
    
    Útil para:
    - Tests unitarios (rápido, no necesita BD)
    - Desarrollo inicial
    - Prototipos
    """

    def __init__(self):
        # Almacenamiento en memoria
        self._datos: Dict[int, Usuario] = {}
        self._siguiente_id = 1

    def guardar(self, usuario: Usuario) -> Usuario:
        """
        Guarda un usuario.
        Si no tiene ID, asigna uno nuevo.
        """
        if usuario.id is None:
            # Es un usuario nuevo, asignar ID
            usuario.id = self._siguiente_id
            self._siguiente_id += 1
        
        # Guardar (crear o actualizar)
        self._datos[usuario.id] = usuario
        return usuario

    def obtener_por_id(self, id: int) -> Optional[Usuario]:
        """Obtiene usuario por ID."""
        return self._datos.get(id)

    def obtener_todos(self) -> List[Usuario]:
        """Obtiene todos los usuarios."""
        return list(self._datos.values())

    def obtener_por_email(self, email: str) -> Optional[Usuario]:
        """
        Busca usuario por email.
        Método específico de este repositorio.
        """
        for usuario in self._datos.values():
            if usuario.email == email:
                return usuario
        return None

    def eliminar(self, id: int) -> bool:
        """Elimina usuario por ID."""
        if id in self._datos:
            del self._datos[id]
            return True
        return False

    def existe(self, id: int) -> bool:
        """Verifica si existe el usuario."""
        return id in self._datos

    def existe_email(self, email: str) -> bool:
        """Verifica si existe un usuario con ese email."""
        return any(u.email == email for u in self._datos.values())

    def limpiar(self):
        """Limpia todos los datos (útil para tests)."""
        self._datos.clear()
        self._siguiente_id = 1

4.3 Implementación con SQLAlchemy (producción)

# infrastructure/repositories/usuario_sqlalchemy_repo.py
from sqlalchemy.orm import Session
from typing import List, Optional
from domain.repositories.base import Repository
from domain.models.usuario import Usuario
from infrastructure.database.models import UsuarioDB  # Modelo SQLAlchemy

class UsuarioSQLAlchemyRepository(Repository[Usuario]):
    """
    Implementación con SQLAlchemy del repositorio.
    
    Usa una base de datos real (PostgreSQL, SQLite, etc.)
    """

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

    def guardar(self, usuario: Usuario) -> Usuario:
        """Guarda usuario en base de datos."""
        if usuario.id is None:
            # Crear nuevo
            db_usuario = UsuarioDB(
                username=usuario.username,
                email=usuario.email,
                nombre_completo=usuario.nombre_completo,
                activo=usuario.activo
            )
            self._session.add(db_usuario)
        else:
            # Actualizar existente
            db_usuario = self._session.query(UsuarioDB).get(usuario.id)
            db_usuario.username = usuario.username
            db_usuario.email = usuario.email
            db_usuario.activo = usuario.activo
        
        self._session.commit()
        
        # Actualizar el ID en la entidad de dominio
        usuario.id = db_usuario.id
        return usuario

    def obtener_por_id(self, id: int) -> Optional[Usuario]:
        """Obtiene usuario de la BD."""
        db_usuario = self._session.query(UsuarioDB).get(id)
        if db_usuario:
            return self._to_domain(db_usuario)
        return None

    def _to_domain(self, db_usuario: UsuarioDB) -> Usuario:
        """Convierte modelo DB a entidad de dominio."""
        return Usuario(
            id=db_usuario.id,
            username=db_usuario.username,
            email=db_usuario.email,
            nombre_completo=db_usuario.nombre_completo,
            activo=db_usuario.activo
        )

    # ... otros métodos ...

5. Inyección de Dependencias

El patrón Repository se combina con Inyección de Dependencias para desacoplar completamente el dominio de la infraestructura.

5.1 Servicio que usa Repository

# application/services/usuario_service.py
from domain.models.usuario import Usuario
from domain.repositories.base import Repository

class UsuarioService:
    """
    Servicio de aplicación para gestión de usuarios.
    
    NO sabe CÓMO se persisten los datos.
    Solo sabe QUE se pueden guardar/obtener mediante el repository.
    """

    def __init__(self, repo: Repository[Usuario]):
        """
        Constructor recibe el repositorio inyectado.
        
        Args:
            repo: Cualquier implementación de Repository[Usuario]
                  (puede ser MemoryRepository para tests
                   o SQLAlchemyRepository para producción)
        """
        self._repo = repo

    def crear_usuario(self, username: str, email: str, 
                      nombre_completo: str) -> Usuario:
        """
        Crea un nuevo usuario con validaciones.
        """
        # Validaciones de negocio
        if len(username) < 3:
            raise ValueError("Username debe tener al menos 3 caracteres")
        
        if self._repo.existe_email(email):
            raise ValueError(f"Ya existe un usuario con email {email}")
        
        # Crear entidad
        usuario = Usuario(
            id=None,  # Sin ID, se asignará al guardar
            username=username,
            email=email,
            nombre_completo=nombre_completo
        )
        
        # Guardar usando el repositorio
        return self._repo.guardar(usuario)

    def obtener_usuario(self, id: int) -> Usuario:
        """
        Obtiene un usuario por ID.
        """
        usuario = self._repo.obtener_por_id(id)
        if not usuario:
            raise ValueError(f"Usuario con ID {id} no encontrado")
        return usuario

    def desactivar_usuario(self, id: int) -> Usuario:
        """
        Desactiva un usuario.
        """
        usuario = self.obtener_usuario(id)
        usuario.desactivar()
        return self._repo.guardar(usuario)

    def listar_usuarios_activos(self) -> List[Usuario]:
        """
        Lista todos los usuarios activos.
        """
        todos = self._repo.obtener_todos()
        return [u for u in todos if u.activo]

5.2 Configuración de Dependencias

# config/dependencies.py
from infrastructure.repositories.usuario_repo import UsuarioMemoryRepository
from infrastructure.repositories.usuario_sqlalchemy_repo import UsuarioSQLAlchemyRepository
from application.services.usuario_service import UsuarioService

def create_usuario_service_for_tests():
    """
    Crea servicio con repositorio en memoria.
    Útil para tests unitarios (rápido, aislado).
    """
    repo = UsuarioMemoryRepository()
    return UsuarioService(repo)

def create_usuario_service_for_production(db_session):
    """
    Crea servicio con repositorio de base de datos.
    Úsalo en la aplicación real.
    """
    repo = UsuarioSQLAlchemyRepository(db_session)
    return UsuarioService(repo)

# Ejemplo de uso
if __name__ == "__main__":
    # Para tests
    service_test = create_usuario_service_for_tests()
    
    # Crear usuarios
    usuario1 = service_test.crear_usuario(
        "juan123", "juan@email.com", "Juan Pérez"
    )
    print(f"Creado: {usuario1}")
    
    # Obtener
    usuario2 = service_test.obtener_usuario(usuario1.id)
    print(f"Obtenido: {usuario2}")
    
    # Desactivar
    usuario3 = service_test.desactivar_usuario(usuario1.id)
    print(f"Activos: {len(service_test.listar_usuarios_activos())}")

5.3 Comparación: Antes vs Después DDD

class UsuarioService:
    def crear_usuario(self, username, email):
        # ¡ACOPLADO a SQLite!
        import sqlite3
        conn = sqlite3.connect('taskflow.db')
        cursor = conn.cursor()
        
        # Lógica mezclada con SQL
        cursor.execute(
            "INSERT INTO usuarios ...",
            (username, email)
        )
        conn.commit()
        conn.close()
        
# Problemas:
# - No se puede testear sin BD real
# - No se puede cambiar a PostgreSQL
# - Lógica de negocio mezclada con SQL
# - Difícil de mantener

class UsuarioService:
    def __init__(self, repo: Repository[Usuario]):
        self._repo = repo  # Inyectado

    def crear_usuario(self, username, email):
        # Solo lógica de negocio
        if len(username) < 3:
            raise ValueError("Username muy corto")
        
        usuario = Usuario(
            id=None,
            username=username,
            email=email
        )
        return self._repo.guardar(usuario)

# Beneficios:
# - Testeable con repo fake
# - Fácil cambiar implementación
# - Lógica limpia y pura
# - Mantenible

Ejercicio: Refactorizar TaskFlow con DDD

Aplica los conceptos de DDD al módulo de Tareas de TaskFlow.

Tareas a Realizar

1. Crear Entidad Tarea:
• Atributos: id, titulo, descripcion, estado, fecha_creacion, fecha_limite
• Métodos: completar(), cambiar_prioridad(), esta_atrasada()
2. Crear Value Objects:
• Prioridad (baja, media, alta, urgente)
• EstadoTarea (pendiente, en_progreso, completada)
3. Crear Repositorio:
• TareaRepository con operaciones CRUD
• Implementación en memoria
4. Crear Servicio:
• TareaService con lógica de negocio
• Inyección del repositorio
5. Tests: Crear tests unitarios para todo

Estructura de Archivos Sugerida

taskflow/
├── domain/
│   ├── models/
│   │   ├── __init__.py
│   │   ├── usuario.py          # Entidad Usuario
│   │   └── tarea.py            # Entidad Tarea
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── email.py
│   │   ├── prioridad.py
│   │   └── estado_tarea.py
│   └── repositories/
│       ├── __init__.py
│       └── base.py             # Interfaz Repository
├── application/
│   ├── __init__.py
│   └── services/
│       ├── __init__.py
│       ├── usuario_service.py
│       └── tarea_service.py
├── infrastructure/
│   ├── repositories/
│   │   ├── __init__.py
│   │   ├── usuario_memory_repo.py
│   │   └── tarea_memory_repo.py
│   └── database/
│       └── ...                 # SQLAlchemy models (más adelante)
└── tests/
    ├── test_usuario.py
    ├── test_tarea.py
    ├── test_usuario_service.py
    └── test_tarea_service.py

Ejemplo de Implementación

# domain/models/tarea.py
from dataclasses import dataclass
from datetime import datetime, date
from typing import Optional
from domain.value_objects.prioridad import Prioridad
from domain.value_objects.estado_tarea import EstadoTarea

@dataclass
class Tarea:
    id: Optional[int]
    titulo: str
    descripcion: str
    prioridad: Prioridad
    estado: EstadoTarea = EstadoTarea("pendiente")
    fecha_creacion: datetime = None
    fecha_limite: Optional[date] = None

    def __post_init__(self):
        if self.fecha_creacion is None:
            self.fecha_creacion = datetime.now()

    def completar(self):
        """Marca la tarea como completada."""
        self.estado = EstadoTarea("completada")

    def esta_atrasada(self) -> bool:
        """Verifica si la tarea está atrasada."""
        if self.fecha_limite is None:
            return False
        return (
            self.estado.valor != "completada" and 
            date.today() > self.fecha_limite
        )

    def cambiar_prioridad(self, nueva_prioridad: Prioridad):
        """Cambia la prioridad de la tarea."""
        self.prioridad = nueva_prioridad

# domain/value_objects/prioridad.py
from dataclasses import dataclass

@dataclass(frozen=True)
class Prioridad:
    valor: str

    def __post_init__(self):
        validas = {"baja", "media", "alta", "urgente"}
        if self.valor not in validas:
            raise ValueError(f"Prioridad inválida: {self.valor}")

# application/services/tarea_service.py
from domain.models.tarea import Tarea
from domain.repositories.base import Repository
from domain.value_objects.prioridad import Prioridad

class TareaService:
    def __init__(self, repo: Repository[Tarea]):
        self._repo = repo

    def crear_tarea(self, titulo: str, descripcion: str, 
                    prioridad_str: str) -> Tarea:
        if len(titulo) < 3:
            raise ValueError("Título muy corto")
        
        prioridad = Prioridad(prioridad_str)
        
        tarea = Tarea(
            id=None,
            titulo=titulo,
            descripcion=descripcion,
            prioridad=prioridad
        )
        
        return self._repo.guardar(tarea)

    def completar_tarea(self, tarea_id: int) -> Tarea:
        tarea = self._repo.obtener_por_id(tarea_id)
        if not tarea:
            raise ValueError("Tarea no encontrada")
        
        tarea.completar()
        return self._repo.guardar(tarea)

    def listar_tareas_pendientes(self):
        todas = self._repo.obtener_todos()
        return [t for t in todas if t.estado.valor == "pendiente"]

Resumen y Checklist