Los middlewares en FastAPI son funciones de Python que se ejecutan justo antes y después de cada request, lo que te permite modificar el comportamiento global de tu API sin tocar cada endpoint. Si trabajas con FastAPI y necesitas validaciones, métricas o logs centralizados, dominar este patrón te ahorra horas de código repetido.
¿Qué es un middleware en FastAPI y para qué sirve?
Un middleware intercepta cada request entrante y cada response saliente. Es el lugar perfecto para aplicar lógica que debe correr en todas las rutas, sin duplicar código.
¿Qué es un middleware? Es una función que se ejecuta antes y después de cada request en tu API, ideal para validaciones globales, métricas y logs.
Un caso típico es el de CORS validations, un middleware que valida si el request proviene de un dominio aceptado por tu aplicación. FastAPI ya trae varios middlewares listos para usar, así que en proyectos reales lo más común es reutilizar los que ya existen en el paquete antes de escribir uno propio.
¿Cuándo conviene crear un custom middleware?
Cuando necesitas una lógica específica que ningún middleware preexistente cubre. En el ejemplo de la clase, construimos uno que mide cuánto tarda en procesarse cada request, una métrica útil para detectar endpoints lentos.
¿Cómo crear un middleware personalizado en FastAPI?
Dentro de tu archivo main.py, justo después de declarar la aplicación, defines una función asíncrona que recibe dos parámetros: el request actual y una función call_next que ejecuta el flujo por defecto.
La estructura mínima se ve así:
python import time from fastapi import Request
@app.middleware("http") async def log_request_time(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time print(f"Request {request.url} completed in {process_time:.4f} seconds") return response
Fíjate en tres detalles importantes:
@app.middleware("http") registra la función como middleware de tipo HTTP.await call_next(request) ejecuta el resto del pipeline y devuelve la respuesta.response, porque estás interceptando el ciclo completo del request.Sin las líneas de start_time y process_time, este middleware sería transparente: dejaría pasar el request tal como llegó. La magia está en lo que agregas alrededor de call_next.
¿Cómo medir el tiempo de procesamiento de un request?
La lógica es directa. Guardas time.time() antes de llamar a call_next y vuelves a capturarlo después. La resta te da los segundos exactos que tardó el endpoint en responder.
¿Para qué sirve
call_nexten un middleware? Es la función que ejecuta el resto del pipeline del request. Sin ella, tu middleware bloquearía la respuesta.
Al ejecutar la app y llamar, por ejemplo, al endpoint de transactions con un límite de 50 y offset de 0, la consola imprime algo como: Request /transactions completed in 0.0123 seconds. Si después llamas a /plans, verás otro tiempo, normalmente menor, porque cada endpoint tiene su propio costo de procesamiento.
¿Qué errores comunes aparecen al implementar middlewares?
Un detalle que aparece en la clase es un conflicto de nombres. Si en clases anteriores definiste una función llamada time, vas a chocar con la clase time de Python que necesitas importar para medir el tiempo.
La solución es renombrar tu función. En el ejemplo, time pasó a llamarse get_time_by_iso_code, lo que liberó el espacio de nombres y permitió que time.time() funcionara correctamente.
Esta lección aplica a cualquier proyecto Python: evita usar nombres que colisionen con módulos estándar como time, os, json o request.
¿Qué información puedes capturar desde el objeto request?
El parámetro request te da acceso a múltiples atributos útiles. Algunos de los más frecuentes son:
request.url para conocer la ruta llamada.request.method para saber si fue GET, POST, PUT u otro.request.headers para inspeccionar los headers enviados por el cliente.request.client para identificar la dirección del consumidor.Esta información te sirve para construir middlewares de auditoría, autenticación ligera o trazabilidad sin modificar tus endpoints.
¿Qué reto puedes resolver para practicar middlewares?
El ejercicio propuesto es claro: crea un nuevo middleware que imprima en la consola todos los headers que llegan a cada endpoint. Es una práctica corta pero te obliga a recorrer request.headers, formatear la salida y entender el orden de ejecución del pipeline.
Cuéntame en los comentarios qué otros middlewares personalizados se te ocurren para tus propios proyectos en FastAPI.
Los middlewares en FastAPI son funciones de Python que se ejecutan justo antes y después de cada request, lo que te permite modificar el comportamiento global de tu API sin tocar cada endpoint. Si trabajas con FastAPI y necesitas validaciones, métricas o logs centralizados, dominar este patrón te ahorra horas de código repetido.
Un middleware intercepta cada request entrante y cada response saliente. Es el lugar perfecto para aplicar lógica que debe correr en todas las rutas, sin duplicar código.
¿Qué es un middleware? Es una función que se ejecuta antes y después de cada request en tu API, ideal para validaciones globales, métricas y logs.
Un caso típico es el de CORS validations, un middleware que valida si el request proviene de un dominio aceptado por tu aplicación. FastAPI ya trae varios middlewares listos para usar, así que en proyectos reales lo más común es reutilizar los que ya existen en el paquete antes de escribir uno propio.
Cuando necesitas una lógica específica que ningún middleware preexistente cubre. En el ejemplo de la clase, construimos uno que mide cuánto tarda en procesarse cada request, una métrica útil para detectar endpoints lentos.
Dentro de tu archivo main.py, justo después de declarar la aplicación, defines una función asíncrona que recibe dos parámetros: el request actual y una función call_next que ejecuta el flujo por defecto.
La estructura mínima se ve así:
python import time from fastapi import Request
@app.middleware("http") async def log_request_time(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time print(f"Request {request.url} completed in {process_time:.4f} seconds") return response
Fíjate en tres detalles importantes:
@app.middleware("http") registra la función como middleware de tipo HTTP.await call_next(request) ejecuta el resto del pipeline y devuelve la respuesta.response, porque estás interceptando el ciclo completo del request.Sin las líneas de start_time y process_time, este middleware sería transparente: dejaría pasar el request tal como llegó. La magia está en lo que agregas alrededor de call_next.
La lógica es directa. Guardas time.time() antes de llamar a call_next y vuelves a capturarlo después. La resta te da los segundos exactos que tardó el endpoint en responder.
¿Para qué sirve
call_nexten un middleware? Es la función que ejecuta el resto del pipeline del request. Sin ella, tu middleware bloquearía la respuesta.
Al ejecutar la app y llamar, por ejemplo, al endpoint de transactions con un límite de 50 y offset de 0, la consola imprime algo como: Request /transactions completed in 0.0123 seconds. Si después llamas a /plans, verás otro tiempo, normalmente menor, porque cada endpoint tiene su propio costo de procesamiento.
Un detalle que aparece en la clase es un conflicto de nombres. Si en clases anteriores definiste una función llamada time, vas a chocar con la clase time de Python que necesitas importar para medir el tiempo.
La solución es renombrar tu función. En el ejemplo, time pasó a llamarse get_time_by_iso_code, lo que liberó el espacio de nombres y permitió que time.time() funcionara correctamente.
Esta lección aplica a cualquier proyecto Python: evita usar nombres que colisionen con módulos estándar como time, os, json o request.
El parámetro request te da acceso a múltiples atributos útiles. Algunos de los más frecuentes son:
request.url para conocer la ruta llamada.request.method para saber si fue GET, POST, PUT u otro.request.headers para inspeccionar los headers enviados por el cliente.request.client para identificar la dirección del consumidor.Esta información te sirve para construir middlewares de auditoría, autenticación ligera o trazabilidad sin modificar tus endpoints.
El ejercicio propuesto es claro: crea un nuevo middleware que imprima en la consola todos los headers que llegan a cada endpoint. Es una práctica corta pero te obliga a recorrer request.headers, formatear la salida y entender el orden de ejecución del pipeline.
Cuéntame en los comentarios qué otros middlewares personalizados se te ocurren para tus propios proyectos en FastAPI.
Solución al reto:
@app.middleware("http")
async def log_request_headers(request: Request, call_next):
print("Request Headers:")
for header, value in request.headers.items():
print(f"{header}: {value}")
response = await call_next(request)
return response
Solución al reto:
@app.middleware("http")
async def log_request_headers(request: Request, call_next) -> Request:
'''
Middleware para registrar los encabezados de la solicitud.
Parámetros:
- request: La solicitud entrante.
'''
print(f"Request headers: {request.headers}")
response = await call_next(request)
return response
solucion:@app.middleware("http")async def log_request_headers(request: Request, call_next): response = await call_next(request) print(f"Request: {request.url} headers: {dict(request.headers)}") return response
@app.middleware("http")
async def log_request_headers(request: Request, call_next):
response = await call_next(request)
print(f"Request: {request.url} headers: {dict(request.headers)}")
return response
@app.middleware("http")
async def log_request_heder(request: <u>Request</u>, call_next):
start_time = <u>time</u>.perf_counter()
# Headers del request
print("\n--- REQUEST HEADERS ---")
for key, value in request.headers.items():
print(f"{key}: {value}")
response = await call_next(request)
process_time = <u>time</u>.perf_counter() - start_time
# Header personalizado
response.headers["X-Process-Time"] = <u>str</u>(process_time)
# Headers del response
print("\n--- RESPONSE HEADERS ---")
for key, value in response.headers.items():
print(f"{key}: {value}")
print(f"\n{request.method} {request.url.path} - {process_time:.4f}s")
return response
@app.middleware("http")
async def print_headers(request: Request, call_next):
print(f"Headers: {request.headers}") response = await call_next(request) return response
Solucion al reto:
Me gusta mas mantener segmentado el codigo y es mejor trabajar con la clase BaseHTTPMiddleware.
Ejemplo:
import time from fastapi import FastAPI, Request, Response, status from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware
class LogHandler(BaseHTTPMiddleware): def __init__(self, app: FastAPI) -> None: super().__init__(app)
async def dispatch(self, request: Request, call_next) -> Response | JSONResponse: try: start_time = time.time() response = await call_next(request) end_time = time.time() process_time = end_time - start_time print(f"Request: {request.url} complete in: {process_time:.4f}") return response except Exception as error: return JSONResponse(status_code=status.STATUS_500, content={'error': str(error)})
Para registar el middlaware, seri solo:
app.add_middleware(LogHandler)
Mi solucion muy simple: @app.middleware("http")async def add_process_time_header(request: Request, call_next): print(request.headers) response = await call_next(request) return response
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
print(request.headers)
response = await call_next(request)
return response
En FastAPI, los middlewares son componentes que se ejecutan antes y después de que cada solicitud sea manejada por un endpoint. Son útiles para tareas como autenticación, registro de solicitudes, manejo de errores, compresión y más.
Un middleware es una función o clase que intercepta las solicitudes HTTP entrantes y las respuestas salientes. Esto permite realizar operaciones personalizadas, como:
En FastAPI, los middlewares se implementan como clases que heredan de BaseHTTPMiddleware o se definen como funciones personalizadas.
from fastapi import FastAPI, Request from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class LogMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # Antes de procesar la solicitud print(f"Request: {request.method} {request.url}") # Procesar la solicitud response = await call_next(request) # Después de procesar la solicitud print(f"Response status: {response.status_code}") return response
# Agregar el middleware a la aplicación app.add_middleware(LogMiddleware)
@app.get("/") async def read_root(): return {"message": "Hello, World!"}
FastAPI permite integrar middlewares ya existentes de Starlette o externos, como:
Permite configurar qué dominios pueden hacer solicitudes a tu API.
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware( CORSMiddleware, allow_origins=["https://example.com"], # Dominios permitidos allow_credentials=True, allow_methods=["*"], # Métodos permitidos (GET, POST, etc.) allow_headers=["*"], # Encabezados permitidos )
@app.get("/") async def read_root(): return {"message": "Hello, CORS!"}
Comprime las respuestas para mejorar el rendimiento.
from starlette.middleware.gzip import GZipMiddleware
app = FastAPI()
app.add_middleware(GZipMiddleware)
@app.get("/") async def read_root(): return {"message": "This response is compressed!"}
Supongamos que deseas validar un encabezado personalizado en cada solicitud.
from fastapi import Request, HTTPException from starlette.middleware.base import BaseHTTPMiddleware
class HeaderValidationMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): if "x-custom-header" not in request.headers: raise HTTPException(status_code=400, detail="Missing x-custom-header") # Pasar al siguiente middleware o endpoint response = await call_next(request) return response
app = FastAPI()
app.add_middleware(HeaderValidationMiddleware)
@app.get("/") async def read_root(): return {"message": "Header is valid!"}
El orden en el que los middlewares son añadidos es importante, ya que se ejecutan en secuencia. Por ejemplo:
app.add_middleware(MiddlewareA) app.add_middleware(MiddlewareB)
En este caso:
MiddlewareA se ejecuta primero al procesar la solicitud.MiddlewareB se ejecuta después al procesar la solicitud.MiddlewareB → MiddlewareA.Aunque los middlewares son globales y afectan todas las solicitudes, las dependencias en FastAPI ofrecen una forma más granular de validar o modificar solicitudes por endpoint.
Usa middlewares para tareas generales y dependencias para validaciones específicas.
Calcula cuánto tiempo tarda cada solicitud en ser procesada.
import time from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware
class TimerMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response
app = FastAPI()
app.add_middleware(TimerMiddleware)
@app.get("/") async def read_root(): return {"message": "Check the X-Process-Time header for timing info."}
Los middlewares en FastAPI son una herramienta poderosa para gestionar solicitudes y respuestas a nivel global. Puedes usarlos para optimizar tu API con tareas como autenticación, registro, compresión y validaciones. Asegúrate de combinarlos sabiamente con dependencias para una arquitectura eficiente.