Middleware en FastAPI

¿Qué es un middleware?

El middleware es un componente de software que se sitúa entre el cliente y los endpoints de tu API, procesando todas las peticiones entrantes y todas las respuestas salientes. Piensa en él como un punto de control universal: cada petición pasa por el middleware antes de llegar a cualquier endpoint, y cada respuesta pasa por él antes de ser enviada al cliente.

Esta capacidad lo convierte en el lugar ideal para implementar funcionalidades transversales, aquellas que afectan a todos o a muchos endpoints sin que sea necesario duplicar código en cada uno de ellos.

Ciclo de vida de una petición con middleware

Cuando un cliente envía una petición a tu API, el flujo es:

Cliente → Middleware (antes) → Endpoint → Middleware (después) → Cliente

El middleware puede:

• Leer y modificar la petición entrante
• Decidir si continuar procesando o devolver una respuesta directamente
• Leer y modificar la respuesta antes de enviarla al cliente

Creando el primer middleware

En FastAPI, el middleware HTTP más sencillo se define con el decorador @app.middleware("http"). La función recibe dos parámetros: la petición (request) y una función callable (call_next) que representa al resto de la cadena de procesamiento.

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def mi_primer_middleware(request: Request, call_next):
    print(f"Petición recibida: {request.method} {request.url}")
    respuesta = await call_next(request)
    print(f"Respuesta enviada: {respuesta.status_code}")
    return respuesta

@app.get("/")
async def raiz():
    return {"mensaje": "Hola, mundo"}

El código antes de await call_next(request) se ejecuta antes de que la petición llegue al endpoint. El código después se ejecuta una vez que el endpoint ya ha generado la respuesta, pero antes de que sea enviada al cliente.

Estructura completa de un middleware

from fastapi import FastAPI, Request
from fastapi.responses import Response

app = FastAPI()

@app.middleware("http")
async def middleware_completo(request: Request, call_next):
    # --- Antes del endpoint ---
    # Aquí puedes leer headers, URL, método, etc.
    metodo = request.method
    ruta = request.url.path

    # Llama al siguiente paso (endpoint u otro middleware)
    respuesta = await call_next(request)

    # --- Después del endpoint ---
    # Aquí puedes modificar la respuesta
    respuesta.headers["X-Mi-Cabecera"] = "valor-personalizado"

    return respuesta

Casos de uso prácticos

Medir el tiempo de respuesta

Un caso de uso muy habitual es registrar cuánto tarda cada endpoint en responder. Esto ayuda a detectar cuellos de botella en producción.

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def medir_tiempo(request: Request, call_next):
    tiempo_inicio = time.perf_counter()

    respuesta = await call_next(request)

    duracion_ms = (time.perf_counter() - tiempo_inicio) * 1000
    respuesta.headers["X-Process-Time-Ms"] = f"{duracion_ms:.2f}"

    print(f"{request.method} {request.url.path} → {respuesta.status_code} ({duracion_ms:.2f}ms)")

    return respuesta

Cuando ejecutes tu API y hagas peticiones, verás en la consola algo como:

GET /usuarios → 200 (12.45ms)
POST /productos → 201 (34.21ms)

Registrar todas las peticiones

Para auditoría y depuración, es útil registrar cada petición con su método, ruta, código de estado y tiempo:

import time
import logging
from fastapi import FastAPI, Request

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def registrar_peticiones(request: Request, call_next):
    inicio = time.perf_counter()

    respuesta = await call_next(request)

    duracion = time.perf_counter() - inicio

    logger.info(
        "method=%s path=%s status=%d duration=%.3fs ip=%s",
        request.method,
        request.url.path,
        respuesta.status_code,
        duracion,
        request.client.host if request.client else "desconocido"
    )

    return respuesta

Añadir cabeceras de seguridad globales

Las cabeceras de seguridad HTTP son una buena práctica para proteger tu API. En lugar de añadirlas endpoint por endpoint, el middleware las aplica a todas las respuestas:

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def cabeceras_seguridad(request: Request, call_next):
    respuesta = await call_next(request)

    # Evita que el navegador "adivine" el tipo de contenido
    respuesta.headers["X-Content-Type-Options"] = "nosniff"
    # Protección básica contra clickjacking
    respuesta.headers["X-Frame-Options"] = "DENY"
    # Fuerza HTTPS en navegadores que lo soporten
    respuesta.headers["Strict-Transport-Security"] = "max-age=31536000"

    return respuesta

ID de correlación por petición

En sistemas distribuidos, es útil asignar un identificador único a cada petición para rastrearla a través de todos los servicios:

import uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def agregar_id_correlacion(request: Request, call_next):
    # Usa el ID proporcionado por el cliente, o genera uno nuevo
    correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))

    respuesta = await call_next(request)

    # Devuelve el mismo ID en la respuesta para que el cliente pueda rastrearlo
    respuesta.headers["X-Correlation-ID"] = correlation_id

    return respuesta

Múltiples middleware: orden de ejecución

Puedes añadir varios middleware a tu aplicación. Se ejecutan en orden inverso al de su declaración (el último declarado envuelve al primero):

from fastapi import FastAPI, Request
import time
import uuid

app = FastAPI()

@app.middleware("http")
async def middleware_seguridad(request: Request, call_next):
    respuesta = await call_next(request)
    respuesta.headers["X-Content-Type-Options"] = "nosniff"
    return respuesta

@app.middleware("http")
async def middleware_tiempo(request: Request, call_next):
    inicio = time.perf_counter()
    respuesta = await call_next(request)
    duracion = (time.perf_counter() - inicio) * 1000
    respuesta.headers["X-Process-Time"] = f"{duracion:.2f}ms"
    return respuesta

@app.middleware("http")
async def middleware_correlacion(request: Request, call_next):
    correlation_id = str(uuid.uuid4())
    respuesta = await call_next(request)
    respuesta.headers["X-Correlation-ID"] = correlation_id
    return respuesta

Middleware con BaseHTTPMiddleware

Para middleware más complejos, Starlette ofrece la clase BaseHTTPMiddleware que permite una organización más limpia:

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import FastAPI, Request

class MiddlewarePersonalizado(BaseHTTPMiddleware):
    def __init__(self, app, prefijo_cabecera: str = "X-App"):
        super().__init__(app)
        self.prefijo_cabecera = prefijo_cabecera

    async def dispatch(self, request: Request, call_next):
        respuesta = await call_next(request)
        respuesta.headers[f"{self.prefijo_cabecera}-Version"] = "1.0"
        return respuesta

app = FastAPI()
app.add_middleware(MiddlewarePersonalizado, prefijo_cabecera="X-MiAPI")

Esta forma es útil cuando el middleware necesita configuración o mantener estado.

Middleware vs dependencias

Es importante distinguir cuándo usar middleware y cuándo usar el sistema de dependencias:

| Característica | Middleware | Dependencias | |---|---|---| | Ámbito | Todas las peticiones | Endpoints específicos | | Acceso al response | Sí | No (excepto con yield) | | Acceso al body | Complejo | Sencillo | | Inyección de datos al endpoint | No directamente | Sí | | Casos de uso típicos | Logging, cabeceras, CORS | Auth, validación, base de datos |

Regla general: si la lógica afecta a todos los endpoints y necesitas modificar la respuesta, usa middleware. Si la lógica es específica de ciertos endpoints y necesitas inyectar datos en ellos, usa dependencias.

Middleware integrado en Starlette

Además de crear middleware personalizado, puedes usar los middleware listos para usar que incluye Starlette:

from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

# Comprimir respuestas grandes automáticamente
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Solo aceptar peticiones a hosts conocidos
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["midominio.com", "*.midominio.com", "localhost"]
)

# En producción, redirigir HTTP a HTTPS
# app.add_middleware(HTTPSRedirectMiddleware)

El GZipMiddleware es especialmente útil para APIs que devuelven respuestas grandes como listados de datos o exportaciones, ya que puede reducir el tamaño de la respuesta hasta un 70%.