FastAPI en producción: lo que aprendí después de 8 meses y 2M de requests

FastAPI se volvió mi framework Python de referencia para APIs después de años con Flask y Django REST Framework. Es rápido, tiene validación automática con Pydantic y la documentación OpenAPI automática ahorra horas. Pero después de 8 meses con él en producción atendiendo millones de requests, aprendí que el tutorial te lleva al 20% del camino. Aquí está el otro 80%.

La estructura que uso en proyectos reales

El patrón que me ha funcionado mejor escala bien y mantiene separación de responsabilidades:

app/
├── main.py              # Entry point, lifespan, middleware
├── config.py            # Settings con pydantic-settings
├── dependencies.py      # Inyección de dependencias compartidas
├── routers/
│   ├── users.py
│   └── orders.py
├── models/
│   ├── user.py          # Modelos SQLAlchemy
│   └── order.py
├── schemas/
│   ├── user.py          # Pydantic schemas (request/response)
│   └── order.py
├── services/
│   ├── user_service.py  # Lógica de negocio
│   └── order_service.py
└── repositories/
    ├── user_repo.py     # Queries a DB
    └── order_repo.py

La separación entre schemas (Pydantic) y models (SQLAlchemy) es crítica. Mezclarlos es el error más común que veo en proyectos FastAPI.

Lifespan: el reemplazo correcto de startup/shutdown

Desde FastAPI 0.93, los eventos startup y shutdown están deprecados. El patrón correcto usa el context manager lifespan:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import engine, Base
from app.cache import redis_client

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    await redis_client.ping()
    print("✅ Base de datos y caché conectados")

    yield  # La aplicación corre aquí

    # Shutdown
    await redis_client.close()
    await engine.dispose()
    print("🔌 Conexiones cerradas correctamente")

app = FastAPI(lifespan=lifespan)

Dependency Injection: el arma secreta de FastAPI

La inyección de dependencias de FastAPI no es solo para la DB. Úsala para autenticación, rate limiting y cualquier cosa que múltiples endpoints necesiten:

from fastapi import Depends, HTTPException, Header
from app.services.auth import verify_token, get_current_user

async def require_auth(authorization: str = Header(...)) -> dict:
    """Dependencia reutilizable de autenticación."""
    token = authorization.replace("Bearer ", "")
    payload = verify_token(token)
    if not payload:
        raise HTTPException(status_code=401, detail="Token inválido")
    return payload

async def require_admin(user = Depends(require_auth)) -> dict:
    """Dependencia que extiende require_auth."""
    if user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Acceso denegado")
    return user

# Uso en routers
@router.get("/admin/users", dependencies=[Depends(require_admin)])
async def list_all_users(db: AsyncSession = Depends(get_db)):
    ...

El encadenamiento de dependencias (require_admin depende de require_auth) es elegante y completamente testeable.

SQLAlchemy async: el patrón correcto

El error más frecuente con FastAPI + SQLAlchemy async es crear una sesión por request sin cerrarla correctamente:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # Verifica la conexión antes de usarla
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False  # Evita lazy loading después del commit
)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

expire_on_commit=False evita el MissingGreenlet error que aparece cuando intentas acceder a atributos de un objeto después de hacer commit en un contexto async.

Middleware de logging estructurado

En producción necesitas saber qué pasó. Este middleware registra cada request con datos que realmente sirven:

import time
import uuid
import logging
from fastapi import Request

logger = logging.getLogger("api.access")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()

    response = await call_next(request)

    duration_ms = (time.perf_counter() - start) * 1000
    logger.info(
        "request",
        extra={
            "request_id": request_id,
            "method": request.method,
            "path": request.url.path,
            "status": response.status_code,
            "duration_ms": round(duration_ms, 2),
            "client_ip": request.client.host,
        }
    )
    return response

Con este logging estructurado puedes hacer queries en CloudWatch, Datadog o cualquier sistema de observabilidad sin parsear strings.

El número que más importa en producción

Después de 8 meses, el bottleneck más común no fue FastAPI: fue el pool de conexiones a la base de datos. Con pool_size=5 (el default), una spike de tráfico causa timeouts en cascada. Dimensiona tu pool según tus workers: pool_size = workers * 2 es un buen punto de partida.

FastAPI es el framework. La infraestructura alrededor es el producto.