🔎 Buscar

🚀 FastAPI

Instalación, parámetros Path y Query, modelos Pydantic con validación, dependencias con Depends, routers, middlewares, autenticación JWT con OAuth2, TestClient, endpoints async, WebSockets y deployment con uvicorn y Docker.

Wiki / Apuntes📖 Contenido

FastAPI

FastAPI combina velocidad de desarrollo y garantías de tipado. Está construido sobre Starlette (HTTP) y Pydantic (validación), y es async por diseño. Al declarar los tipos de tus datos, genera automáticamente la documentación OpenAPI y una UI interactiva en /docs. Este artículo va desde el primer endpoint hasta una API de autenticación completa.

Instalación y primer endpoint

python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install "fastapi[standard]" uvicorn             # incluye uvicorn y httpx
# main.py
from fastapi import FastAPI
app = FastAPI(title="Mi API")
@app.get("/")
def raiz():
    return {"mensaje": "hola mundo"}
@app.get("/saludo/{nombre}")
def saludo(nombre: str):
    return {"saludo": f"Hola, {nombre}"}
uvicorn main:app --reload --port 8000   # main:app = archivo, objeto; --reload solo en dev

Abre http://127.0.0.1:8000/docs para la UI de Swagger; los tipos declarados se reflejan en la documentación.

Path, Query y Body parameters

Origen Cómo lo declaras Ejemplo en la URL
Path argumento en la ruta /usuarios/42
Query con valor por defecto /buscar?q=python&limite=5
Body un modelo Pydantic cuerpo JSON
from fastapi import FastAPI, Path, Query
app = FastAPI()
@app.get("/usuarios/{usuario_id}")
def leer_usuario(
    usuario_id: int = Path(ge=1),
    detalle: bool = Query(False),
    q: str | None = Query(None, max_length=50),
):
    return {"usuario_id": usuario_id, "detalle": detalle, "q": q}
  • Path(ge=1) valida usuario_id >= 1; Query(False) da valor por defecto; str | None hace q opcional.

⚠️ Los argumentos Path no pueden tener valor por defecto. Decláralos primero o usa Path(...).

Modelos Pydantic: validación

Declaras tipos y restricciones; FastAPI valida cada request y devuelve 422 con los errores si algo no cuadra:

from pydantic import BaseModel, Field, EmailStr
class UsuarioIn(BaseModel):
    nombre: str = Field(min_length=2, max_length=50)
    email: EmailStr
    edad: int = Field(ge=0, le=150, default=18)   # default: valor si falta
    activo: bool = True
@app.post("/usuarios")
def crear_usuario(usuario: UsuarioIn):
    return {"recibido": usuario.model_dump()}
  • Field(ge=..., le=...): restricciones numéricas; Field(min_length=...): de longitud. EmailStr requiere pip install email-validator.

model_config

from pydantic import BaseModel, ConfigDict
class Producto(BaseModel):
    model_config = ConfigDict(
        extra="forbid",            # rechaza campos no declarados
        str_strip_whitespace=True, # recorta espacios
        frozen=True,               # inmutable
    )
    nombre: str
    precio: float
Opción Efecto
extra="forbid" error si llega un campo no declarado
extra="ignore" ignora campos extra (por defecto)
str_strip_whitespace recorta espacios
frozen=True modelo inmutable
from_attributes=True crear desde objetos (ORM)

Dependencias con Depends

from fastapi import FastAPI, Depends
app = FastAPI()
def get_db():
    db = {"conectada": True}      # en una app real: abrir conexión
    try:
        yield db                  # se inyecta en el endpoint
    finally:
        db["conectada"] = False   # cleanup al terminar
@app.get("/datos")
def leer_datos(db=Depends(get_db)):
    return {"db": db}

💡 Las dependencias con yield (dependencies with yield) ejecutan el teardown aunque el endpoint falle. El patrón típico con SQLAlchemy: get_db abre SessionLocal(), hace yield db y cierra en finally.

Sub-dependencias

def get_current_user(token: str = Depends(oauth2_scheme)):
    return verificar_token(token)
def get_activo(user=Depends(get_current_user)):
    if not user["activo"]:
        raise HTTPException(status_code=403, detail="usuario inactivo")
    return user
@app.get("/perfil")
def perfil(user=Depends(get_activo)):
    return user

Routers: modulariza tu app

# app/routers/usuarios.py
from fastapi import APIRouter
router = APIRouter(prefix="/usuarios", tags=["usuarios"])
@router.get("/")
def listar():
    return [{"id": 1}, {"id": 2}]
@router.post("/")
def crear(usuario: UsuarioIn):
    return usuario
# app/main.py
from fastapi import FastAPI
from app.routers import usuarios
app = FastAPI()
app.include_router(usuarios.router)   # prefix cuelga todas las rutas; tags las agrupa en docs

Middlewares

import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def medir_tiempo(request: Request, call_next):
    inicio = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Procesado-ms"] = str(round((time.perf_counter() - inicio) * 1000, 2))
    return response

CORS

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_methods=["*"],
    allow_headers=["*"],
)

⚠️ En producción, allow_origins=["*"] es inseguro si la API usa cookies o credenciales. Lista solo los orígenes que confías.

Autenticación: OAuth2PasswordBearer + JWT

Hashear contraseñas con passlib y bcrypt

pip install "passlib[bcrypt]" python-jose[cryptography]
# app/security.py
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)
def verificar_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

⚠️ Nunca guardes contraseñas en texto plano. bcrypt añade una sal aleatoria, así que cada usuario tiene un hash distinto aunque repita contraseña.

Generar y verificar tokens JWT con python-jose

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
SECRET_KEY = "cambia-esta-clave-en-produccion"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def crear_token(subject: str) -> str:
    expira = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({"sub": subject, "exp": expira}, SECRET_KEY, algorithm=ALGORITHM)
def decodificar_token(token: str) -> str:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])["sub"]
    except JWTError:
        raise ValueError("token inválido o expirado")

El esquema OAuth2 y el login

OAuth2PasswordBearer declara dónde esperar el token (Authorization: Bearer ...):

# app/auth.py
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from app.security import crear_token, verificar_password, decodificar_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
USUARIOS = {"ana": {"username": "ana", "password_hash": None}}
def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        username = decodificar_token(token)
    except ValueError:
        raise HTTPException(status_code=401, detail="token inválido")
    usuario = USUARIOS.get(username)
    if usuario is None:
        raise HTTPException(status_code=401, detail="usuario no encontrado")
    return usuario
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    usuario = USUARIOS.get(form.username)
    if not usuario or not verificar_password(form.password, usuario["password_hash"]):
        raise HTTPException(status_code=401, detail="credenciales inválidas")
    return {"access_token": crear_token(form.username), "token_type": "bearer"}
@app.get("/usuarios/me")
def leer_mi_usuario(usuario: dict = Depends(get_current_user)):
    return usuario

OAuth2PasswordRequestForm espera el body como application/x-www-form-urlencoded con username y password, el formato estándar de OAuth2 para login.

Testing con TestClient

TestClient envuelve la app y permite hacer requests sin levantar un servidor real (usa httpx):

from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_raiz():
    r = client.get("/")
    assert r.status_code == 200
    assert r.json() == {"mensaje": "hola mundo"}

💡 TestClient como context manager (with TestClient(app) as c:) es necesario cuando tu app tiene eventos startup/shutdown o dependencias con yield.

Endpoints async

FastAPI distingue def (síncrona, thread pool) de async def (event loop). La regla de oro:

from fastapi import FastAPI
import httpx
app = FastAPI()
# async def: ideal para I/O async (httpx, drivers async)
@app.get("/github/{repo}")
async def datos_github(repo: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.github.com/repos/{repo}")
    return r.json()
# def: ideal para código síncrono (SQLAlchemy, requests)
@app.get("/sincrono")
def sincrono():
    return {"nota": "FastAPI lo corre en un thread pool"}

WebSockets

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        mensaje = await websocket.receive_text()
        await websocket.send_text(f"eco: {mensaje}")

⚠️ Los WebSockets no se prueban con TestClient igual que las requests HTTP; usa el client de httpx con client.websocket_connect.

Deployment: uvicorn y Docker

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4   # un proceso por núcleo

⚠️ --workers no funciona con --reload. Con varios workers el estado en memoria no se comparte: guarda sesiones y cachés en servicios externos (Redis, DB).

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
docker build -t mi-api . && docker run -p 8000:8000 mi-api

💡 Usa imágenes -slim, combina COPY + RUN pip install para la caché de Docker, y no corras la app como root.

Ejemplo completo: API de autenticación con JWT y base de datos

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
    pass
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
# app/models.py
from sqlalchemy import Column, Integer, String
from app.database import Base
class Usuario(Base):
    __tablename__ = "usuarios"
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    password_hash = Column(String)

app/security.py reutiliza el módulo de la sección de autenticación (hash_password, verificar_password, crear_token, decodificar_token) con una SECRET_KEY propia.

# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from pydantic import BaseModel, EmailStr
from app.database import get_db, Base, engine
from app.models import Usuario
from app.security import hash_password, verificar_password, crear_token, decodificar_token
Base.metadata.create_all(bind=engine)
app = FastAPI(title="API Auth")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class UsuarioRegistro(BaseModel):
    username: str
    email: EmailStr
    password: str
def get_current_user(token=Depends(oauth2_scheme), db: Session = Depends(get_db)):
    try:
        username = decodificar_token(token)
    except ValueError:
        raise HTTPException(status_code=401, detail="token inválido")
    usuario = db.query(Usuario).filter(Usuario.username == username).first()
    if usuario is None:
        raise HTTPException(status_code=401, detail="usuario no encontrado")
    return usuario
@app.post("/registro")
def registro(datos: UsuarioRegistro, db: Session = Depends(get_db)):
    if db.query(Usuario).filter(Usuario.username == datos.username).first():
        raise HTTPException(status_code=400, detail="usuario ya existe")
    usuario = Usuario(username=datos.username, email=datos.email,
                      password_hash=hash_password(datos.password))
    db.add(usuario)
    db.commit()
    return {"mensaje": "usuario creado"}
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    usuario = db.query(Usuario).filter(Usuario.username == form.username).first()
    if not usuario or not verificar_password(form.password, usuario.password_hash):
        raise HTTPException(status_code=401, detail="credenciales inválidas")
    return {"access_token": crear_token(usuario.username), "token_type": "bearer"}
@app.get("/usuarios/me")
def perfil(usuario: Usuario = Depends(get_current_user)):
    return {"id": usuario.id, "username": usuario.username, "email": usuario.email}

Puedes probar el flujo completo (registro → login → perfil) con TestClient, como vimos en la sección de testing.

Para profundizar

Estudio · Recursos de todo el mundo (inglés, chino, japonés, español, francés, ruso…) curados y traducidos al español.