Cómo construir una API REST asíncrona y escalable en Python con FastAPI, SQLModel y Docker

Tutorial práctico paso a paso para crear una API REST asíncrona y preparada para producción usando FastAPI, SQLModel (SQLAlchemy + Pydantic), Docker y buenas prácticas: estructura de carpetas, ejemplos completos de código, pruebas básicas y consejos de rendimiento y seguridad.

Por qué esta pila

• FastAPI: rendimiento, validación automática y documentación OpenAPI.
• SQLModel: modelo unificado entre Pydantic y SQLAlchemy, facilita la serialización y el ORM.
• Async: mejor uso de I/O para manejar muchas conexiones concurrentes.
• Docker: despliegue consistente y reproducible.

Estructura de proyecto

project/
├─ app/
│  ├─ main.py
│  ├─ api.py
│  ├─ db.py
│  ├─ models.py
│  ├─ crud.py
│  ├─ schemas.py
│  ├─ deps.py
│  └─ config.py
├─ tests/
│  └─ test_basic.py
├─ Dockerfile
├─ docker-compose.yml
├─ requirements.txt
└─ README.md

requirements.txt

fastapi
uvicorn[standard]
sqlmodel
alembic
asyncpg            # si usas PostgreSQL
databases[postgresql]
httpx
pytest
python-dotenv

Config básica (app/config.py)

from pydantic import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str = 'sqlite+aiosqlite:///./test.db'
    APP_NAME: str = 'fastapi-sqlmodel-app'
    DEBUG: bool = True

    class Config:
        env_file = '.env'

settings = Settings()

Conexión DB asíncrona (app/db.py)

from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from .config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=settings.DEBUG, future=True)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

Modelos (app/models.py)

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime

class ItemBase(SQLModel):
    name: str
    description: Optional[str] = None

class Item(ItemBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)

class ItemCreate(ItemBase):
    pass

class ItemRead(ItemBase):
    id: int
    created_at: datetime

CRUD (app/crud.py)

from sqlmodel import select
from .models import Item, ItemCreate
from .db import async_session
from sqlmodel.ext.asyncio.session import AsyncSession

async def create_item(item_in: ItemCreate) -> Item:
    async with async_session() as session:  # type: AsyncSession
        item = Item.from_orm(item_in)
        session.add(item)
        await session.commit()
        await session.refresh(item)
        return item

async def get_item(item_id: int) -> Item | None:
    async with async_session() as session:
        result = await session.get(Item, item_id)
        return result

async def list_items(offset: int = 0, limit: int = 50):
    async with async_session() as session:
        stmt = select(Item).offset(offset).limit(limit)
        result = await session.exec(stmt)
        return result.all()

Dependencias y seguridad (app/deps.py)

from fastapi import Depends, HTTPException, status

async def get_pagination_params(skip: int = 0, limit: int = 50):
    if limit > 100:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='limit too high')
    return {'skip': skip, 'limit': limit}

# Para autenticación, aquí pondrías un get_current_user con OAuth2/JWT

Rutas (app/api.py)

from fastapi import APIRouter, Depends, HTTPException
from .schemas import ItemCreate, ItemRead
from .crud import create_item, get_item, list_items
from .deps import get_pagination_params

router = APIRouter(prefix='/items', tags=['items'])

@router.post('/', response_model=ItemRead)
async def post_item(item: ItemCreate):
    return await create_item(item)

@router.get('/{item_id}', response_model=ItemRead)
async def read_item(item_id: int):
    item = await get_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail='Item not found')
    return item

@router.get('/', response_model=list[ItemRead])
async def read_items(p: dict = Depends(get_pagination_params)):
    return await list_items(offset=p['skip'], limit=p['limit'])

Entrypoint (app/main.py)

from fastapi import FastAPI
from .api import router as api_router
from .db import init_db
from .config import settings

app = FastAPI(title=settings.APP_NAME)
app.include_router(api_router)

@app.on_event('startup')
async def on_startup():
    await init_db()

Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .\nCMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80", "--proxy-headers"]

docker-compose.yml (ejemplo con PostgreSQL)

version: '3.8'
services:
  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: appdb
    volumes:
      - db-data:/var/lib/postgresql/data
  web:
    build: .
    environment:
      DATABASE_URL: 'postgresql+asyncpg://user:pass@db:5432/appdb'
    depends_on:
      - db
    ports:
      - '8000:80'
volumes:
  db-data:

Testing básico (tests/test_basic.py)

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_root():
    async with AsyncClient(app=app, base_url='http://test') as ac:
        r = await ac.get('/items/')
        assert r.status_code == 200

Por qué se hace así (decisiones importantes)

• Usar SQLModel permite definir modelos Pydantic/ORM en un solo lugar, reduciendo duplicación.
• async_sessionmaker + AsyncSession: evita bloqueos del event loop y mejora concurrencia bajo carga I/O-bound.
• Separación en módulos (crud, deps, api) mejora testabilidad y mantiene responsabilidades claras.
• Commit y refresh se hacen dentro del contexto de sesión para mantener consistencia de transacción.

Mejoras y consideraciones para producción

• Usar Uvicorn con Gunicorn (uvicorn.workers.UvicornWorker) para múltiples procesos.
• Configurar migrations con Alembic + SQLModel: genera y aplica migrations en CI/CD.
• Usar pooling y tuning de la base de datos: parámetros de pool_size, max_overflow según DB.
• Agregar autenticación (OAuth2 + JWT), permisos y rate limiting (e.g., via Redis).
• Monitoreo: métricas (Prometheus), tracing (OpenTelemetry) y logs estructurados.

Seguridad rápida

• Validación estricta de entrada gracias a Pydantic/SQLModel.
• Evita exponer información sensible en errores; maneja excepciones globalmente.
• Usa HTTPS en producción y encabezados HSTS, CORS configurado con origenes permitidos.
• Protege endpoints críticos con autenticación y scopes.

Consejos de rendimiento

• Prefiere consultas selectivas (proyecciones) en lugar de cargar objetos completos si no son necesarios.
• Usa índices adecuados en la base de datos para consultas frecuentes.
• Cachea respuestas inmutables o caras con Redis o Varnish.
• Evita N+1 queries: usa joins o cargado explícito en consultas complejas.

Próximos pasos recomendados

• Agregar pruebas de integración que levanten una DB en memoria o contenedor.
• Implementar CI/CD: ejecutar linters, tests y migraciones antes del despliegue.
• Configurar un gateway (NGINX) y balanceo de carga para múltiples réplicas del backend.

Tip avanzado: para cargas muy altas, combina procesos (Gunicorn) y workers asíncronos (UvicornWorker), optimiza el pool de conexiones async y mide latencias con counters para identificar cuellos de botella en la base de datos.

Advertencia: no ignores las migraciones de esquema en producción: aplicar cambios sin migraciones probadas puede corromper datos; siempre probar en staging y tener backups.