IF0100 - Programación OO II

Objetivos de Aprendizaje

Al finalizar esta clase, serás capaz de:

• Entender el patrón Repository y su propósito
• Crear una interfaz genérica de repositorio
• Implementar repositorios con SQLAlchemy ORM (Object-Relational Mapping)
• Inyectar repositorios en FastAPI
• Desacoplar la lógica de negocio de la persistencia

Que es el Patron Repository? (15 min)

El Repository Pattern abstrae el acceso a datos, simulando una coleccion en memoria.

Capas de la Arquitectura

┌─────────────────────────────────────┐
│           PRESENTACION              │  (FastAPI routes, Jinja2)
├─────────────────────────────────────┤
│            SERVICIOS                │  (Logica de negocio)
├─────────────────────────────────────┤
│          REPOSITORIOS               │  (Acceso a datos)
├─────────────────────────────────────┤
│           MODELOS                   │  (Entidades de dominio)
├─────────────────────────────────────┤
│         BASE DE DATOS               │  (PostgreSQL, SQLite)
└─────────────────────────────────────┘

Interfaz Generica (20 min)

# repositories/base.py
from abc import ABC, abstractmethod
from typing import TypeVar, Generic, Optional, List

T = TypeVar("T")

class Repositorio(ABC, Generic[T]):
    """Interfaz generica para repositorios."""
    
    @abstractmethod
    def obtener_por_id(self, id: int) -> Optional[T]:
        """Obtener entidad por ID."""
        pass
    
    @abstractmethod
    def obtener_todos(self) -> List[T]:
        """Obtener todas las entidades."""
        pass
    
    @abstractmethod
    def crear(self, entidad: T) -> T:
        """Crear nueva entidad."""
        pass
    
    @abstractmethod
    def actualizar(self, entidad: T) -> T:
        """Actualizar entidad existente."""
        pass
    
    @abstractmethod
    def eliminar(self, id: int) -> bool:
        """Eliminar entidad por ID."""
        pass
    
    @abstractmethod
    def existe(self, id: int) -> bool:
        """Verificar si existe entidad."""
        pass
    
    @abstractmethod
    def contar(self) -> int:
        """Contar entidades."""
        pass

Implementacion SQLAlchemy (25 min)

# repositories/usuario_repo.py
from typing import Optional, List
from sqlalchemy.orm import Session
from repositories.base import Repositorio
from models.usuario import Usuario

class UsuarioRepositorioSQLAlchemy(Repositorio[Usuario]):
    """Implementacion de repositorio para Usuario con SQLAlchemy."""
    
    def __init__(self, db: Session):
        self.db = db
    
    def obtener_por_id(self, id: int) -> Optional[Usuario]:
        return self.db.query(Usuario).filter(Usuario.id == id).first()
    
    def obtener_por_username(self, username: str) -> Optional[Usuario]:
        return self.db.query(Usuario).filter(Usuario.username == username).first()
    
    def obtener_todos(self) -> List[Usuario]:
        return self.db.query(Usuario).all()
    
    def obtener_activos(self) -> List[Usuario]:
        return self.db.query(Usuario).filter(Usuario.activo == True).all()
    
    def crear(self, usuario: Usuario) -> Usuario:
        self.db.add(usuario)
        self.db.commit()
        self.db.refresh(usuario)
        return usuario
    
    def actualizar(self, usuario: Usuario) -> Usuario:
        self.db.commit()
        self.db.refresh(usuario)
        return usuario
    
    def eliminar(self, id: int) -> bool:
        usuario = self.obtener_por_id(id)
        if usuario:
            self.db.delete(usuario)
            self.db.commit()
            return True
        return False
    
    def existe(self, id: int) -> bool:
        return self.db.query(Usuario).filter(Usuario.id == id).count() > 0
    
    def contar(self) -> int:
        return self.db.query(Usuario).count()


# Repositorio en memoria para testing
class UsuarioRepositorioMemoria(Repositorio[Usuario]):
    """Implementacion en memoria para pruebas."""
    
    def __init__(self):
        self._datos: dict[int, Usuario] = {}
        self._next_id = 1
    
    def obtener_por_id(self, id: int) -> Optional[Usuario]:
        return self._datos.get(id)
    
    def obtener_todos(self) -> List[Usuario]:
        return list(self._datos.values())
    
    def crear(self, usuario: Usuario) -> Usuario:
        usuario.id = self._next_id
        self._next_id += 1
        self._datos[usuario.id] = usuario
        return usuario
    
    def actualizar(self, usuario: Usuario) -> Usuario:
        if usuario.id in self._datos:
            self._datos[usuario.id] = usuario
        return usuario
    
    def eliminar(self, id: int) -> bool:
        if id in self._datos:
            del self._datos[id]
            return True
        return False
    
    def existe(self, id: int) -> bool:
        return id in self._datos
    
    def contar(self) -> int:
        return len(self._datos)

Inyeccion en FastAPI (15 min)

# api/dependencies.py
from typing import Generator
from sqlalchemy.orm import Session
from fastapi import Depends
from database import SessionLocal
from repositories.usuario_repo import UsuarioRepositorioSQLAlchemy

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_usuario_repo(db: Session = Depends(get_db)) -> UsuarioRepositorioSQLAlchemy:
    return UsuarioRepositorioSQLAlchemy(db)


# api/routes/usuarios.py
from fastapi import APIRouter, Depends, HTTPException
from repositories.usuario_repo import UsuarioRepositorioSQLAlchemy
from api.dependencies import get_usuario_repo

router = APIRouter(prefix="/api/usuarios", tags=["usuarios"])

@router.get("/{id}")
def obtener_usuario(
    id: int,
    repo: UsuarioRepositorioSQLAlchemy = Depends(get_usuario_repo)
):
    usuario = repo.obtener_por_id(id)
    if not usuario:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return usuario

@router.get("/")
def listar_usuarios(
    repo: UsuarioRepositorioSQLAlchemy = Depends(get_usuario_repo)
):
    return repo.obtener_todos()

Ejercicio: Repositorio Tarea (15 min)

Crear TareaRepositorioSQLAlchemy con metodos adicionales:

# repositories/tarea_repo.py
class TareaRepositorioSQLAlchemy(Repositorio[Tarea]):
    
    def obtener_por_proyecto(self, proyecto_id: int) -> List[Tarea]:
        """Obtener tareas de un proyecto."""
        return self.db.query(Tarea).filter(
            Tarea.proyecto_id == proyecto_id
        ).all()
    
    def obtener_completadas(self, proyecto_id: int) -> List[Tarea]:
        """Obtener tareas completadas."""
        return self.db.query(Tarea).filter(
            Tarea.proyecto_id == proyecto_id,
            Tarea.completada == True
        ).all()
    
    def obtener_pendientes(self, proyecto_id: int) -> List[Tarea]:
        """Obtener tareas pendientes."""
        return self.db.query(Tarea).filter(
            Tarea.proyecto_id == proyecto_id,
            Tarea.completada == False
        ).all()
    
    def marcar_completada(self, id: int) -> bool:
        """Marcar tarea como completada."""
        tarea = self.obtener_por_id(id)
        if tarea:
            tarea.completada = True
            self.db.commit()
            return True
        return False

Resumen

• Repository: Abstrae acceso a datos como coleccion en memoria
• Interfaz generica: Define operaciones CRUD comunes
• Desacoplamiento: Logica de negocio no conoce implementacion
• Testing: Repositorios en memoria para pruebas
• Dependency Injection: FastAPI inyecta repositorios