IF0100 - Programacion OO II

Objetivos de Aprendizaje

Al finalizar esta clase, seras capaz de:

• Implementar paginacion en endpoints
• Crear filtros y busqueda avanzada
• Manejar errores HTTP correctamente
• Implementar versionado de API
• Documentar API con OpenAPI

Paginacion (20 min)

from fastapi import Query
from typing import Generic, TypeVar, List
from pydantic import BaseModel

T = TypeVar("T")

class Paginado(BaseModel, Generic[T]):
    items: List[T]
    total: int
    pagina: int
    por_pagina: int
    total_paginas: int

def calcular_offset(pagina: int, por_pagina: int) -> int:
    return (pagina - 1) * por_pagina

@router.get("/tareas", response_model=Paginado[TareaResponse])
def listar_tareas(
    pagina: int = Query(1, ge=1),
    por_pagina: int = Query(10, ge=1, le=100),
    repo: TareaRepositorio = Depends(get_tarea_repo)
):
    # Calcular offset
    offset = calcular_offset(pagina, por_pagina)
    
    # Obtener datos
    tareas = repo.obtener_todos(skip=offset, limit=por_pagina)
    total = repo.contar()
    
    # Calcular total paginas
    total_paginas = (total + por_pagina - 1) // por_pagina
    
    return Paginado(
        items=[TareaResponse.from_entity(t) for t in tareas],
        total=total,
        pagina=pagina,
        por_pagina=por_pagina,
        total_paginas=total_paginas
    )

Filtros y Busqueda (20 min)

from fastapi import Query
from typing import Optional
from enum import Enum
from datetime import datetime

class OrdenDireccion(str, Enum):
    ASC = "asc"
    DESC = "desc"

class TareaFiltro(BaseModel):
    """Filtros para tareas."""
    completada: Optional[bool] = None
    prioridad: Optional[Prioridad] = None
    proyecto_id: Optional[int] = None
    buscar: Optional[str] = None
    fecha_desde: Optional[datetime] = None
    fecha_hasta: Optional[datetime] = None
    ordenar_por: str = "creado_en"
    orden_dir: OrdenDireccion = OrdenDireccion.DESC

@router.get("/tareas")
def listar_tareas(
    completada: Optional[bool] = Query(None),
    prioridad: Optional[Prioridad] = Query(None),
    proyecto_id: Optional[int] = Query(None),
    buscar: Optional[str] = Query(None, min_length=2),
    pagina: int = Query(1, ge=1),
    por_pagina: int = Query(10, ge=1, le=100),
    repo: TareaRepositorio = Depends(get_tarea_repo)
):
    # Construir query
    query = repo.db.query(Tarea)
    
    # Aplicar filtros
    if completada is not None:
        query = query.filter(Tarea.completada == completada)
    
    if prioridad:
        query = query.filter(Tarea.prioridad == prioridad)
    
    if proyecto_id:
        query = query.filter(Tarea.proyecto_id == proyecto_id)
    
    if buscar:
        query = query.filter(
            Tarea.titulo.ilike(f"%{buscar}%") |
            Tarea.descripcion.ilike(f"%{buscar}%")
        )
    
    # Contar total
    total = query.count()
    
    # Aplicar paginacion
    offset = calcular_offset(pagina, por_pagina)
    tareas = query.offset(offset).limit(por_pagina).all()
    
    return construir_respuesta_paginada(tareas, total, pagina, por_pagina)

Manejo de Errores (15 min)

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

# Excepciones personalizadas
class AppException(Exception):
    def __init__(self, status_code: int, mensaje: str, detalle: dict = None):
        self.status_code = status_code
        self.mensaje = mensaje
        self.detalle = detalle or {}

class NoEncontradoException(AppException):
    def __init__(self, recurso: str, id: int):
        super().__init__(
            status_code=404,
            mensaje=f"{recurso} no encontrado",
            detalle={"id": id}
        )

class ConflictoException(AppException):
    def __init__(self, mensaje: str):
        super().__init__(status_code=409, mensaje=mensaje)

# Handler global
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": True,
            "mensaje": exc.mensaje,
            "detalle": exc.detalle
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": True,
            "mensaje": "Datos de entrada invalidos",
            "errores": exc.errors()
        }
    )

# Uso en endpoints
@router.get("/tareas/{id}")
def obtener_tarea(id: int, repo: TareaRepositorio = Depends(get_tarea_repo)):
    tarea = repo.obtener_por_id(id)
    if not tarea:
        raise NoEncontradoException("Tarea", id)
    return TareaResponse.from_entity(tarea)

Versionado de API (15 min)

from fastapi import APIRouter

# Estrategia: URL path versioning

# v1
router_v1 = APIRouter(prefix="/api/v1")

@router_v1.get("/tareas")
def listar_tareas_v1():
    # Version 1: Lista simple
    return {"tareas": []}

# v2
router_v2 = APIRouter(prefix="/api/v2")

@router_v2.get("/tareas")
def listar_tareas_v2(
    pagina: int = 1,
    por_pagina: int = 10
):
    # Version 2: Con paginacion
    return {
        "items": [],
        "total": 0,
        "pagina": pagina,
        "por_pagina": por_pagina
    }

# Registrar en app
app.include_router(router_v1)
app.include_router(router_v2)

# Otra opcion: Header versioning
@app.middleware("http")
async def version_middleware(request: Request, call_next):
    version = request.headers.get("X-API-Version", "v1")
    request.state.api_version = version
    return await call_next(request)

Ejercicio: API Completa (15 min)

Implementar endpoint completo para Tareas:

• GET /tareas - Listar con paginacion y filtros
• GET /tareas/{id} - Obtener por ID
• POST /tareas - Crear
• PATCH /tareas/{id} - Actualizar
• DELETE /tareas/{id} - Eliminar

Resumen

• Paginacion: Limitar resultados con offset/limit
• Filtros: Parametros Query para filtrar
• Errores: HTTPException y handlers globales
• Versionado: URL path o header versioning
• Documentacion: OpenAPI automatico de FastAPI