Concepto clave

La validación avanzada y manejo de errores en FastAPI para APIs de ML en producción es como el sistema de control de calidad en una fábrica automotriz. No solo se trata de rechazar piezas defectuosas, sino de identificar exactamente qué falló, registrar el problema para mejorar procesos futuros y mantener la línea de producción funcionando sin interrupciones. En el contexto de ML, esto significa asegurar que las predicciones sean confiables incluso cuando los datos de entrada no sean perfectos.

FastAPI utiliza Pydantic para validación de datos, que funciona como un inspector riguroso que verifica cada campo antes de permitir que los datos entren al modelo. El manejo de errores, por otro lado, es el protocolo de emergencia que se activa cuando algo sale mal, asegurando que la API responda de manera predecible y útil en lugar de fallar silenciosamente o exponer información sensible.

Cómo funciona en la práctica

Imagina que tienes un modelo que predice precios de viviendas. Los usuarios envían datos como área, número de habitaciones y ubicación. Sin validación, podrían enviar un área negativa o una ubicación que no existe, lo que causaría errores en el modelo o predicciones absurdas.

Paso 1: Definir un modelo Pydantic con restricciones específicas para cada campo.

Paso 2: Configurar FastAPI para usar este modelo en los endpoints.

Paso 3: Implementar manejadores de excepciones personalizados que capturen errores comunes y devuelvan respuestas estandarizadas.

Paso 4: Agregar logging detallado para rastrear errores y métricas de validación.

Codigo en accion

Antes: Endpoint sin validación robusta

from fastapi import FastAPI
import numpy as np

app = FastAPI()

# Modelo ML simulado
def predict_price(area: float, rooms: int, location: str) -> float:
    # Lógica simplificada
    return area * 1000 + rooms * 50000

@app.post("/predict")
async def predict(area: float, rooms: int, location: str):
    price = predict_price(area, rooms, location)
    return {"predicted_price": price}

Después: Con validación y manejo de errores

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field, validator
from typing import Optional
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

# Modelo Pydantic para validación
class PredictionRequest(BaseModel):
    area: float = Field(..., gt=0, description="Área en metros cuadrados, debe ser positiva")
    rooms: int = Field(..., ge=1, le=10, description="Número de habitaciones entre 1 y 10")
    location: str = Field(..., min_length=3, max_length=50)
    
    @validator('location')
    def location_must_be_valid(cls, v):
        valid_locations = ["centro", "norte", "sur", "este", "oeste"]
        if v.lower() not in valid_locations:
            raise ValueError(f"Ubicación '{v}' no válida. Opciones: {valid_locations}")
        return v.lower()

# Modelo ML simulado
def predict_price(area: float, rooms: int, location: str) -> float:
    # Lógica con coeficientes por ubicación
    coefficients = {"centro": 1.5, "norte": 1.2, "sur": 1.0, "este": 1.1, "oeste": 1.1}
    base_price = area * 1000 + rooms * 50000
    return base_price * coefficients.get(location, 1.0)

# Manejador de excepciones personalizado
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    logger.error(f"Error de validación: {exc}")
    return JSONResponse(
        status_code=422,
        content={"detail": str(exc), "error_type": "validation_error"}
    )

@app.post("/predict")
async def predict(request: PredictionRequest):
    try:
        price = predict_price(request.area, request.rooms, request.location)
        logger.info(f"Predicción exitosa: {request.dict()}")
        return {
            "predicted_price": round(price, 2),
            "currency": "USD",
            "metadata": {"model_version": "1.0"}
        }
    except Exception as e:
        logger.error(f"Error en predicción: {e}")
        raise HTTPException(status_code=500, detail="Error interno del servidor")

Errores comunes

Validación demasiado permisiva: No definir restricciones suficientes en los modelos Pydantic, permitiendo datos que causan errores en el modelo ML. Solución: Usar Field con parámetros como gt, ge, lt, le para rangos numéricos y validadores personalizados para lógica compleja.
Mensajes de error genéricos: Devolver "error interno" sin detalles útiles para debugging. Solución: Implementar logging estructurado y respuestas de error que incluyan tipo de error y contexto sin exponer información sensible.
Falta de manejo de excepciones específicas: Capturar solo Exception genérico, perdiendo la capacidad de responder diferentemente a distintos tipos de errores. Solución: Usar @app.exception_handler para tipos específicos como ValueError, HTTPException, o excepciones personalizadas.
No validar salidas del modelo: Asumir que el modelo ML siempre devuelve valores válidos. Solución: Agregar validación post-predicción para detectar NaN, infinitos o valores fuera de rango esperado.
Ignorar el contexto de producción: No considerar límites de tasa, timeouts o fallos de dependencias. Solución: Integrar manejo de errores con herramientas como circuit breakers y retries.

Checklist de dominio

¿Defino modelos Pydantic con restricciones específicas para todos los campos de entrada?
¿Implemento manejadores de excepciones personalizados para errores comunes?
¿Uso logging estructurado para rastrear errores y métricas de validación?
¿Valido tanto entradas como salidas del modelo ML?
¿Proporciono mensajes de error útiles pero seguros para el cliente?
¿Documento los códigos de error y formatos de respuesta en la API?
¿Pruebo escenarios de error como datos inválidos, timeouts y fallos del modelo?

Implementa validación avanzada para un endpoint de clasificación de imágenes

En este ejercicio, mejorarás un endpoint FastAPI existente que clasifica imágenes. El endpoint actual acepta una imagen y devuelve una etiqueta, pero tiene problemas de validación.

Contexto inicial: Tienes un archivo main.py con este código base:

from fastapi import FastAPI, File, UploadFile
from PIL import Image
import io

app = FastAPI()

# Modelo simulado
def classify_image(image_bytes):
    # Simulación: siempre devuelve "gato"
    return {"label": "gato", "confidence": 0.95}

@app.post("/classify")
async def classify(file: UploadFile = File(...)):
    contents = await file.read()
    result = classify_image(contents)
    return result

Paso 1: Validación de archivo: Modifica el endpoint para validar que el archivo sea una imagen (JPEG o PNG) y que no exceda 5MB. Usa Pydantic si es posible.
Paso 2: Manejo de errores: Agrega un manejador de excepciones para ValueError que devuelva un código 422 con detalles cuando la validación falle.
Paso 3: Validación de salida: Asegúrate de que la función classify_image siempre devuelva un diccionario con las claves label (string) y confidence (float entre 0 y 1). Agrega validación post-predicción.
Paso 4: Logging: Implementa logging básico que registre intentos de clasificación exitosos y fallidos, incluyendo el nombre del archivo y tamaño.
Paso 5: Prueba: Crea un script de prueba que envíe una imagen válida, una imagen muy grande y un archivo que no sea imagen, verificando las respuestas.

Entrega el código completo del archivo main.py modificado y el script de prueba.

Pistas

Usa el parámetro `max_size` en `File` para limitar el tamaño, pero también valida programáticamente.
Para validar el tipo de archivo, puedes verificar el `content_type` o usar la librería `magic` o `PIL` para intentar abrir la imagen.
Considera crear una excepción personalizada para errores de validación de imagen y manejarla específicamente.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.

Quiz: Validación y Manejo de Errores

Quiz Interactivo