green snake
Photo by Pixabay on Pexels.com

Guía completa de optimización de rendimiento en FastAPI: crea APIs escalables con Async I/O, connection pools, caching y rate limiting


Resumen (comprende primero la visión general)

  • El rendimiento de FastAPI depende no solo del código de tu aplicación, sino de la combinación de ajustes de Uvicorn/Gunicorn, pools de conexiones a la base de datos, caching, limitación de velocidad (rate limiting) y almacenamiento externo.
  • Usando Async I/O correctamente y derivando el trabajo bloqueante a threads o tareas en segundo plano, puedes manejar alta concurrencia.
  • Para bases de datos y APIs externas, utiliza pools de conexiones y caching para evitar repetir cálculos pesados o recuperación de datos.
  • Protege el sistema de picos bruscos de tráfico con rate limiting y colas, equilibrando entre “proteger el sistema” y “ser razonablemente generoso con los clientes”.
  • Finalmente, resumimos una hoja de ruta de medición y ajuste para ver claramente por dónde empezar.

A quién le resultará útil esta guía (perfiles concretos)

  1. Desarrolladores individuales / aprendices

    • Has creado una pequeña app en FastAPI y te preocupa qué se ralentizará cuando aumenten los usuarios.
    • Has oído que “async lo hace más rápido”, pero no sabes exactamente qué cuidados debes tener.
  2. Ingenieros backend en equipos pequeños

    • Operas FastAPI en producción y empiezas a notar peores tiempos de respuesta o timeouts en horas pico.
    • Buscas una visión estructurada del diseño de pools de conexiones, caching y rate limiting.
  3. Equipos SaaS en fase de crecimiento

    • Quieres definir la dirección de una arquitectura escalable antes de que aumente mucho el tráfico.
    • Necesitas criterios para equilibrar rendimiento, fiabilidad y coste.

1. Decide qué quieres acelerar primero

Antes de optimizar a ciegas, aclara “qué quieres mejorar y cómo”. Eso reduce esfuerzo desperdiciado.

1.1 Métricas típicas

  • Latencia: tiempo de respuesta por petición
  • Throughput: peticiones procesadas por segundo (RPS)
  • Conexiones concurrentes
  • Ratio de errores

En la práctica, lo que afecta directamente a la UX es la latencia P95 y el porcentaje de errores.

1.2 Clasifica aproximadamente dónde está el cuello de botella

En FastAPI, los cuellos de botella suelen caer en:

  1. Código de aplicación (cálculos CPU-heavy, conversión JSON, algoritmos no óptimos)
  2. I/O externa (bases de datos, APIs externas, storage, email, etc.)
  3. Configuración de infraestructura (número de workers, pods, límites de conexión, timeouts)

Esta guía explica mentalidad y ejemplos para mejorar 1–3 en equilibrio.


2. Async I/O vs threads / procesos

FastAPI está diseñado sobre Async I/O (async / await). Usado correctamente, es muy potente ante alta concurrencia. Pero no significa “si todo es async, todo será más rápido”.

2.1 Conceptos básicos de funciones async

Ejemplo:

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/weather")
async def get_weather(city: str):
    async with httpx.AsyncClient(timeout=2.0) as client:
        r = await client.get(f"https://api.example.com/weather?city={city}")
    return r.json()

Puntos clave:

  • Usa clientes asincrónicos como httpx.AsyncClient.
  • Mientras esperas I/O, el event loop atiende otras peticiones.

2.2 Trabajo CPU-heavy → threads o background jobs

El trabajo intensivo en CPU bloquea el event loop aunque sea async. Usa:

  • run_in_threadpool
  • Sistemas de background jobs (Celery, etc.)

Ejemplo:

from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool

app = FastAPI()

def heavy_calc(n: int) -> int:
    s = 0
    for i in range(n):
        s += i * i
    return s

@app.get("/heavy")
async def heavy_endpoint(n: int = 100_000_000):
    result = await run_in_threadpool(heavy_calc, n)
    return {"n": n, "result": result}

2.3 Elegir cantidad de workers Uvicorn / Gunicorn

Regla típica:

  • Workers: #CPU × 2
  • Requests concurrentes por worker: muy alto si tu carga es mayormente I/O-async

Pero siempre debes ajustar tras pruebas de carga.


3. Diseño de pools de conexión a la base de datos (SQLAlchemy)

Gran parte del cuello de botella suele estar en el RDB.

3.1 Qué es un connection pool

Reutiliza conexiones en vez de abrir una por petición.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+psycopg://user:pass@db:5432/app"

engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

3.2 Cómo pensar en los valores del pool

  • pool_size: conexiones listas por instancia
  • max_overflow: conexiones adicionales temporales
  • Garantiza no superar max_connections del DB

Ejemplo:

  • 3 instancias × (10 + 10) = 60 conexiones posibles → configura el DB por encima de 60 + margen.

3.3 Gestión correcta de sesiones por petición

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

4. Estrategias de caching: in-memory, Redis, HTTP

4.1 Dónde cachear

  1. En memoria local
  2. Redis u otro cache externo
  3. Caché HTTP (ETag, Cache-Control)

4.2 In-memory cache

from functools import lru_cache
import httpx

@lru_cache(maxsize=128)
def fetch_remote_config():
    r = httpx.get("https://config.example.com/app-config")
    return r.json()

4.3 Cache con Redis

class RedisCache:
    ...
    async def get_json(self, key): ...
    async def set_json(self, key, value, ttl): ...

Patrón “cache → miss → compute → set → return”.

4.4 Caché HTTP

@app.get("/static-data")
def static_data():
    headers = {"Cache-Control": "public, max-age=300"}
    return Response(..., headers=headers)

5. Proteger tu servicio con rate limiting

5.1 Por qué rate limiting

  • Parar bots / clientes defectuosos
  • Proteger durante picos repentinos
  • Gestión de planes gratuitos / cuotas

5.2 Token bucket (explicación simple)

  • Cubo con tokens
  • Cada petición consume 1
  • Se recargan a ritmo constante
  • Si no hay tokens → 429

5.3 Ejemplo simple (in-memory)

class SimpleRateLimiter:
    ...
global_limiter = SimpleRateLimiter(10, 5)

def limit_global():
    if not global_limiter.allow():
        raise HTTPException(429, "Too many requests")

6. No sirvas archivos grandes directamente desde FastAPI

Usa almacenamiento externo + URLs firmadas.

def generate_signed_url(object_key, expires_in=300):
    ...

7. Medir y perfilar: nunca dependas de la intuición

7.1 Middleware de tiempos

class TimingMiddleware(BaseHTTPMiddleware):
    ...

7.2 Herramientas de carga

  • locust, k6, wrk
  • Mide P95, prueba distintas configuraciones de workers, pools, caché, etc.

8. Mini API ejemplo con cache + rate limiting

Incluye:

  • RedisCache
  • RedisRateLimiter
  • Endpoint /expensive con TTL y limitación 30 req/min por IP

9. Errores comunes y soluciones

Síntoma Causa Solución
Spikes de latencia I/O bloqueante, pool pequeño Uso async correcto, ajustar pool, caching
Gran variancia Endpoints pesados o N+1 Optimizar queries, prefetch
DB llega al límite (#instancias × pool_size) grande Ajustar valores y max_connections
Sistema cae en picos Sin rate limiting Limitar, usar colas
Descargas lentas Sirviendo archivos desde API CDN + URLs firmadas
Optimización no se nota Sin métricas Medir antes/después

10. Hoja de ruta de despliegue (por dónde empezar)

  1. Añade middleware de tiempos.
  2. Identifica endpoints lentos, optimiza I/O externa.
  3. Añade caching y mide ratios de cache hit.
  4. Introduce rate limiting.
  5. Externaliza archivos y tareas pesadas.
  6. Pruebas de carga periódicas ajustando configuración.

Resumen

  • El rendimiento de FastAPI depende del conjunto: Async I/O, workers, connection pools, caching, rate limiting y almacenamiento externo.
  • Comienza midiendo; luego ataca los puntos de mayor impacto.
  • Pools, caching y rate limiting son patrones reutilizables.
  • No busques perfección inmediata: mejora paso a paso para crear un servicio rápido, estable y fácil de operar.

por greeden

Deja una respuesta

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

日本語が含まれない投稿は無視されますのでご注意ください。(スパム対策)