Por qué FastAPI para producción
FastAPI combina tipado estricto con Pydantic, documentación automática con Swagger y rendimiento cercano a Go o Node.js gracias a Starlette y uvicorn. Es nuestra elección estándar en Visionary AI Group para APIs de alto tráfico.
Arquitectura de capas
Estructuramos cada proyecto en capas claras:
- Routers: Endpoints HTTP con validación automática vía Pydantic
- Services: Lógica de negocio pura, testeable sin dependencias de infraestructura
- Repositories: Acceso a datos con SQLAlchemy 2.0 async
- Models: Modelos SQLAlchemy para ORM + schemas Pydantic para request/response
python
Estructura de directorios
app/
api/v1/ # Routers
services/ # Lógica de negocio
repositories/ # Acceso a datos
models/ # SQLAlchemy models
schemas/ # Pydantic schemas
core/ # Config, database, security
`
Conexión pool con PostgreSQL
Configuración óptima para 10k+ usuarios concurrentes:
`python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_timeout=30,
pool_recycle=3600,
pool_pre_ping=True, # Verifica conexiones antes de usarlas
)
`
El
pool_pre_ping=True es crítico en producción — detecta conexiones caídas sin lanzar errores al usuario.
Redis como capa de caching
Para endpoints de alta frecuencia, Redis con TTL dinámico reduce la carga en PostgreSQL un 70%:
`python
import redis.asyncio as aioredis
from functools import wraps
async def cache(key: str, ttl: int = 300):
redis = app.state.redis
cached = await redis.get(key)
if cached:
return json.loads(cached)
return None
`
Patrones críticos en producción
Repository pattern con soft delete
Nunca borres datos en producción. Implementa soft delete con
deleted_at:
`python
class BaseRepository:
async def soft_delete(self, obj):
obj.deleted_at = datetime.utcnow()
await self.db.flush()
return obj
`
Rate limiting por IP
`python
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
client_ip = request.client.host
key = f"rate:{client_ip}"
count = await redis.incr(key)
if count == 1:
await redis.expire(key, 60)
if count > 100: # 100 req/min por IP
return JSONResponse(status_code=429, content={"detail": "Rate limit exceeded"})
return await call_next(request)
``
Conclusión
FastAPI + PostgreSQL + Redis es un stack de producción probado que escala horizontalmente sin complicaciones. Con esta arquitectura hemos soportado picos de 50k requests/hora sin degradar la latencia por encima de 200ms.
¿Te fue útil?
Compártelo con tu equipo
Favio Villanueva
CEO & Arquitecto de Software
Arquitecto de software con +8 años de experiencia construyendo sistemas SaaS escalables. Fundador de Visionary AI Group, operando en 12 países.