FastAPI: tipos, validación automática y OpenAPI

Cuando declaras def get_user(id: int) en Flask, el int es decorativo — Flask te pasa un string y tú lo conviertes a mano. FastAPI hace algo cualitativamente diferente: lee las anotaciones de tipo en tiempo de importación, construye un esquema JSON Schema a partir de ellas, valida cada request entrante contra ese esquema, convierte los datos al tipo Python correcto, y genera documentación OpenAPI interactiva, todo antes de que tu función reciba un solo argumento.

El motor detrás de esto es Pydantic v2. FastAPI delega en Pydantic la validación y serialización; tú declaras modelos como subclases de BaseModel con anotaciones de tipo estándar, y Pydantic compila validadores en Rust (vía pydantic-core) que son sustancialmente más rápidos que cualquier validación manual en Python puro. FastAPI también está construido sobre Starlette, que le da soporte nativo de async/await — lo que significa que un worker puede manejar miles de requests concurrentes sin hilos adicionales, siempre que uses I/O asíncrono.

¿Cuándo conviene FastAPI sobre alternativas como Flask o Django REST? Cuando el contrato de la API importa (equipos, clientes externos, mobile), cuando necesitas validación robusta sin librerías extra, o cuando la carga implica mucho I/O concurrente (bases de datos async, llamadas HTTP externas). Si ya tienes un proyecto Django maduro, migrar no tiene sentido. Si empiezas desde cero una API, FastAPI es la opción que elimina más fricción.

Lo que rompe cuando te equivocas: olvidar que Pydantic v2 es estricto con los tipos. Si declaras id: int y llega "abc", FastAPI devuelve un 422 con un JSON detallado explicando exactamente qué campo falló y por qué. Eso es bueno en producción, pero confunde si no lo esperas durante el desarrollo. También, si mezclas rutas síncronas (def) con operaciones de I/O bloqueante, FastAPI las ejecuta en un threadpool, pero si mezclas async def con código bloqueante (como psycopg2 clásico), congelas el event loop completo.

# Requiere: pip install fastapi uvicorn[standard] pydantic sqlalchemy aiosqlite
from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, status
from pydantic import BaseModel, EmailStr, field_validator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# ── Modelos de base de datos (SQLAlchemy async) ─────────────────────────────

DATABASE_URL = "sqlite+aiosqlite:///./users.db"
engine = create_async_engine(DATABASE_URL, echo=False)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


class UserORM(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[str] = mapped_column(unique=True, nullable=False)
    role: Mapped[str] = mapped_column(default="viewer")


# ── Esquemas Pydantic (contrato público de la API) ──────────────────────────

class UserCreate(BaseModel):
    name: str
    email: EmailStr          # Pydantic valida formato de email sin regex manual
    role: str = "viewer"

    @field_validator("role")
    @classmethod
    def role_must_be_valid(cls, v: str) -> str:
        allowed = {"viewer", "editor", "admin"}
        if v not in allowed:
            raise ValueError(f"role debe ser uno de {allowed}")
        return v


class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    role: str

    # Sin esto, Pydantic v2 no sabe cómo leer atributos de un ORM object
    model_config = {"from_attributes": True}


# ── Lifespan: gestión de recursos al arrancar/detener la app ────────────────

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(
    title="User Service",
    version="1.0.0",
    lifespan=lifespan,
)


# ── Dependency: sesión de base de datos ─────────────────────────────────────

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session          # FastAPI cierra la sesión automáticamente al terminar


# Alias tipado para que Depends no repita ruido en cada firma de función
DBSession = Annotated[AsyncSession, Depends(get_db)]


# ── Tarea en background (no bloquea la respuesta al cliente) ─────────────────

async def send_welcome_email(email: str, name: str) -> None:
    # Simula envío asíncrono; en producción llamarías a SendGrid, SES, etc.
    await asyncio.sleep(0.1)
    print(f"[email] Bienvenido {name} → {email}")


# ── Rutas ────────────────────────────────────────────────────────────────────

@app.post(
    "/users",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
)
async def create_user(
    payload: UserCreate,           # FastAPI deserializa y valida el body JSON
    db: DBSession,                 # Inyectado por Depends(get_db)
    background_tasks: BackgroundTasks,
) -> UserResponse:
    from sqlalchemy import select

    # Verificar duplicado antes de insertar
    existing = await db.execute(
        select(UserORM).where(UserORM.email == payload.email)
    )
    if existing.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"El email '{payload.email}' ya está registrado.",
        )

    user = UserORM(**payload.model_dump())
    db.add(user)
    await db.commit()
    await db.refresh(user)

    # El email se envía después de que la respuesta llegue al cliente
    background_tasks.add_task(send_welcome_email, user.email, user.name)

    return UserResponse.model_validate(user)


@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,    # FastAPI convierte el segmento de ruta a int y valida
    db: DBSession,
) -> UserResponse:
    from sqlalchemy import select

    result = await db.execute(select(UserORM).where(UserORM.id == user_id))
    user = result.scalar_one_or_none()

    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Usuario no encontrado.")

    return UserResponse.model_validate(user)

Lo que está pasando realmente

El decorador @app.post("/users", response_model=UserResponse) hace tres cosas simultáneamente: registra la ruta, le dice a FastAPI qué modelo usar para serializar la respuesta (y filtrar campos que no estén en UserResponse, incluso si el objeto tiene más), y añade esa ruta al esquema OpenAPI. Cuando arrancas con uvicorn main:app --reload y abres /docs, ves un Swagger UI completamente funcional generado a partir de esa metadata — sin una línea de configuración YAML.

payload: UserCreate en la firma de create_user desencadena algo preciso: FastAPI detecta que UserCreate es una subclase de BaseModel, asume que viene del cuerpo de la request, lo deserializa y lo valida. Si el body llega sin name, o con un email malformado, o con role: "superadmin", el cliente recibe un 422 con un JSON estructurado que señala exactamente el campo problemático — esto ocurre antes de que tu función se ejecute.

DBSession = Annotated[AsyncSession, Depends(get_db)] es el patrón idiomático para dependency injection en FastAPI. Depends actúa como un mini-contenedor IoC: FastAPI resuelve el grafo de dependencias antes de llamar al handler, y las dependencias que usan yield se comportan como context managers — el código después del yield en get_db corre cuando la request termina, independientemente de si hubo excepción o no. Esto es limpio y testeable: en tests puedes sobrescribir get_db con app.dependency_overrides[get_db] = mock_db.

BackgroundTasks resuelve un problema real: el cliente no debe esperar a que envíes un email de bienvenida. Añades la tarea con background_tasks.add_task(...) y FastAPI la ejecuta en el mismo event loop, después de que la respuesta HTTP ya fue enviada. Para tareas más pesadas o que necesitan reintentos, querrías Celery o ARQ, pero para side effects ligeros BackgroundTasks es exactamente la herramienta correcta.

model_config = {"from_attributes": True} en UserResponse habilita que Pydantic lea atributos desde objetos ORM (en lugar de solo diccionarios). Sin esto, UserResponse.model_validate(user) lanzaría un error porque user es una instancia de UserORM, no un dict.

Errores que debes conocer

Error: mezclar async def con I/O bloqueante congela el event loop entero, afectando todos los requests concurrentes.

# ❌ Wrong
@app.get("/data")
async def get_data(db: DBSession) -> dict:
    import time
    time.sleep(2)  # Bloquea el event loop; nadie más puede procesar requests
    return {"ok": True}

# ✅ Right
import asyncio

@app.get("/data")
async def get_data(db: DBSession) -> dict:
    await asyncio.sleep(2)  # Cede el control; otros requests siguen avanzando
    return {"ok": True}

Si tienes código bloqueante inevitable (una librería sin soporte async), usa def en lugar de async def — FastAPI lo ejecutará en un threadpool automáticamente.


Error: olvidar model_config = {"from_attributes": True} cuando el modelo Pydantic necesita leer desde un ORM object.

# ❌ Wrong
class UserResponse(BaseModel):
    id: int
    name: str

# Lanza: PydanticUserError o ValidationError al intentar validar un ORM object
user_response = UserResponse.model_validate(orm_user_instance)

# ✅ Right
class UserResponse(BaseModel):
    id: int
    name: str
    model_config = {"from_attributes": True}

user_response = UserResponse.model_validate(orm_user_instance)  # Funciona

Con from_attributes=True, Pydantic usa getattr en lugar de acceso por clave de diccionario, lo que hace compatibles los objetos ORM con mapeo de columnas.


Error: usar response_model sin entender que filtra campos — exponer datos sensibles en el ORM object no los expone si no están en el modelo de respuesta, pero solo si siempre pasas por response_model.

# ❌ Wrong: devolver el ORM object directamente saltándose response_model
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DBSession):
    user = await db.get(UserORM, user_id)
    return user  # FastAPI intenta serializar el ORM object; resultado impredecible

# ✅ Right: serializar explícitamente al modelo de respuesta declarado
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: DBSession) -> UserResponse:
    user = await db.get(UserORM, user_id)
    return UserResponse.model_validate(user)  # Contrato claro y filtrado garantizado

Declarar response_model no es suficiente si retornas un objeto que FastAPI no sabe serializar — la serialización explícita vía model_validate hace el contrato visible en el código, no solo en el decorador.

179

Dejar un comentario

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

Scroll al inicio