Cheatsheet FastAPI
Framework Python moderno e rápido para APIs com type hints
FastAPI
Instalação e Setup
Instalar FastAPI
pip install fastapi uvicorn[standard] # crear main.py uvicorn main:app --reload
fastapi es el framework. uvicorn es el servidor ASGI. --reload recarga el código automáticamente en desarrollo. El standard incluye websockets y httptools.
Configuración con Settings
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
debug: bool = False
class Config:
env_file = ".env"
settings = Settings()BaseSettings lee variables de entorno automáticamente. env_file carga desde un archivo .env. Los valores con default son opcionales. Ideal para separar la config del código.
Primera Aplicación
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def root():
return {"msg": "¡Hola FastAPI!"}FastAPI() crea la instancia de la aplicación. @app.get() registra un endpoint GET. El diccionario retornado se serializa a JSON automáticamente. Los type hints generan validación automática.
Metadata de la API
app = FastAPI(
title="Mi API",
description="API de gestión de ítems",
version="1.0.0",
contact={"name": "Equipo Dev"},
)title y description aparecen en la documentación. version controla la versión de la API. contact añade info de contacto. Estos metadatos enriquecen el Swagger UI.
Documentación Automática
# Swagger UI: # http://localhost:8000/docs # ReDoc: # http://localhost:8000/redoc # OpenAPI JSON: # http://localhost:8000/openapi.json
FastAPI genera documentación interactiva automáticamente. /docs muestra Swagger UI con pruebas. /redoc muestra ReDoc formateado. /openapi.json es el schema OpenAPI completo.
Tags para Organización
tags_metadata = [
{"name": "users", "description": "Operaciones de usuarios"},
{"name": "items", "description": "Gestión de ítems"},
]
app = FastAPI(openapi_tags=tags_metadata)
@app.get("/users", tags=["users"])
def list_users(): ...tags agrupan endpoints en la documentación. openapi_tags añade descripciones a las tags. Cada endpoint puede tener múltiples tags. Facilita la navegación en el Swagger UI.
Estructura del Proyecto
app/
main.py
models.py
schemas.py
database.py
dependencies.py
routers/
users.py
items.pymain.py es el punto de entrada. schemas.py define modelos Pydantic. models.py define tablas SQLAlchemy. routers/ organiza los endpoints por módulo. dependencies.py centraliza las dependencias.
Entorno Virtual
python -m venv .venv source .venv/bin/activate # Linux/Mac .venv\Scripts\activate # Windows pip install fastapi uvicorn pip freeze > requirements.txt
venv aísla las dependencias del proyecto. activate activa el entorno virtual. pip freeze genera el archivo de dependencias. Usa siempre entornos virtuales por proyecto.
Rotas e Endpoints
Métodos HTTP
@app.get("/items")
def list_items(): ...
@app.post("/items")
def create(): ...
@app.put("/items/{id}")
def update(id: int): ...
@app.delete("/items/{id}")
def delete(id: int): ...Cada decorador mapea un verbo HTTP. GET para lectura, POST para creación. PUT actualiza totalmente, PATCH parcialmente. DELETE elimina recursos.
Códigos de Estado
from fastapi import status
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create(): ...
@app.delete("/items/{id}", status_code=204)
def delete(id: int): ...
# status.HTTP_404_NOT_FOUND
# status.HTTP_403_FORBIDDENstatus_code define el código HTTP de la respuesta. 201 para recursos creados. 204 para eliminación sin cuerpo. El módulo status tiene constantes con nombre para todos los códigos.
Redirect y Status Code
from fastapi.responses import RedirectResponse
@app.get("/old")
def old():
return RedirectResponse(url="/new", status_code=301)
@app.get("/items/{id}", status_code=200)
def get_item(id: int, response: Response):
response.headers["X-Custom"] = "value"
return {"id": id}RedirectResponse con 301 es un redirect permanente. El parámetro Response permite añadir headers custom. Los headers son útiles para rate-limiting y cache-control.
Path Parameters
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"id": user_id}
# /users/42 → user_id = 42
# /users/abc → error 422Los parámetros entre {} en la ruta se capturan. El type-hint int valida y convierte automáticamente. Los valores inválidos retornan error 422 con detalles. No necesita parsing manual.
Respuestas Personalizadas
from fastapi.responses import JSONResponse, RedirectResponse, HTMLResponse
return JSONResponse(content={"error": "No encontrado"}, status_code=404)
return RedirectResponse(url="/login")
return HTMLResponse(content="<h1>Hola</h1>")JSONResponse permite controlar status y headers. RedirectResponse hace redirect. HTMLResponse retorna HTML directo. Por defecto, los diccionarios se convierten en JSON automáticamente.
Query Parameters
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
return items[skip : skip + limit]
# /items?skip=0&limit=5Los parámetros con valor default se convierten en query params. skip y limit son opcionales en la URL. Sin default, serían obligatorios. FastAPI valida los tipos automáticamente.
Response Model
class UserOut(BaseModel):
id: int
name: str
email: str
@app.get("/users/{id}", response_model=UserOut)
def get_user(id: int):
return user_with_password # el password se filtraresponse_model controla los campos retornados. Los campos fuera del modelo se eliminan automáticamente. Protege datos sensibles como passwords. Genera documentación precisa de la respuesta.
APIRouter
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/")
def list_users(): ...
@router.get("/{user_id}")
def get(user_id: int): ...
# main.py:
app.include_router(router)APIRouter organiza las rutas en módulos separados. prefix añade un camino base a todas las rutas. include_router() lo registra en la app principal. Esencial para proyectos grandes.
Route Naming y Docs
@app.get(
"/items",
summary="Listar ítems",
description="Retorna todos los ítems activos con paginación.",
response_description="Lista de ítems",
deprecated=True,
)
def list_items(): ...summary es el título corto en Swagger. description añade detalles anchos. response_description documenta la respuesta. deprecated=True lo marca como obsoleto en la docs.
Parâmetros e Validação
Body con Pydantic
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
active: bool = True
@app.post("/items")
def create(item: Item):
return itemLos parámetros con tipo BaseModel se leen del body JSON. FastAPI valida los campos automáticamente. Los errores retornan 422 con detalles. No necesita parsing manual del request.
Form Data
from fastapi import Form
@app.post("/login")
def login(
username: str = Form(),
password: str = Form(),
):
return {"user": username}Form() lee datos de formularios HTML (application/x-www-form-urlencoded). Requiere python-multipart instalado. No puede combinarse con body JSON en el mismo endpoint.
Enum Parameters
from enum import Enum
class OrderBy(str, Enum):
name = "name"
price = "price"
date = "date"
@app.get("/items")
def list_items(order: OrderBy = OrderBy.name):
return {"ordered_by": order.value}Enum restringe los valores aceptados. Heredar de str permite comparación directa. Los valores inválidos retornan 422. La documentación muestra los valores disponibles como dropdown.
Validación de Query
from fastapi import Query
@app.get("/items")
def list_items(
q: str | None = Query(None, min_length=3, max_length=50),
page: int = Query(1, ge=1),
):
...Query() añade validaciones a los parámetros de query. min_length y max_length limitan strings. ge (greater or equal) valida números. None como default lo hace opcional.
Upload de Archivos
from fastapi import UploadFile, File
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
content = await file.read()
return {
"name": file.filename,
"type": file.content_type,
"size": len(content),
}UploadFile gestiona uploads con streaming. File(...) lo hace obligatorio. filename y content_type dan metadata. read() es async para archivos grandes.
Validación de Path
from fastapi import Path
@app.get("/items/{item_id}")
def get_item(
item_id: int = Path(ge=1, title="ID del Ítem"),
):
...Path() añade constraints a los path parameters. ge=1 garantiza un valor positivo. title aparece en la documentación. Útil para IDs que deben ser siempre positivos.
Múltiples Archivos
from fastapi import UploadFile, File
@app.post("/uploads")
async def uploads(files: list[UploadFile] = File(...)):
return [
{"name": f.filename, "size": f.size}
for f in files
]list[UploadFile] acepta múltiples archivos. El cliente los envía con el mismo nombre de campo. size da el tamaño sin leerlo todo. Ideal para galerías e imports masivos.
Headers y Cookies
from fastapi import Header, Cookie
@app.get("/info")
def info(
user_agent: str = Header(),
x_token: str | None = Header(None),
session: str | None = Cookie(None),
):
return {"agent": user_agent}Header() lee cabeceras HTTP. Los guiones se convierten en underscores automáticamente (x-token → x_token). Cookie() lee cookies del request. Ambos soportan validación y defaults.
Parámetros Opcionales
@app.get("/items")
def list_items(
q: str | None = None,
category: str | None = None,
max_price: float | None = None,
):
results = items
if q:
results = [i for i in results if q in i["name"]]
return resultsstr | None = None hace el parámetro opcional. Sintaxis union type de Python 3.10+. Si se omite en la URL, recibe None. Permite filtros dinámicos sin múltiples endpoints.
Pydantic e Models
Model Básico
from pydantic import BaseModel
class User(BaseModel):
name: str
email: str
age: int | None = None
active: bool = TrueBaseModel define schemas con validación automática. Los type hints definen el tipo esperado. Los campos con default son opcionales. int | None acepta entero o null.
Model Validators
from pydantic import BaseModel, model_validator
class Registration(BaseModel):
password: str
confirm: str
@model_validator(mode="after")
def check_match(self):
if self.password != self.confirm:
raise ValueError("Las passwords no coinciden")
return self@model_validator valida múltiples campos en conjunto. mode="after" se ejecuta tras la validación individual. Accede a self con todos los campos. Ideal para confirmaciones y dependencias entre campos.
Field y Constraints
from pydantic import BaseModel, Field
class Product(BaseModel):
name: str = Field(min_length=1, max_length=200)
price: float = Field(gt=0, description="Precio en euros")
tags: list[str] = []
quantity: int = Field(default=0, ge=0)Field() añade validaciones y metadata. gt=0 garantiza mayor que cero. min_length valida el tamaño de strings. description aparece en la documentación OpenAPI.
Serialización
user = User(name="Ana", email="ana@site.com") # a diccionario: data = user.model_dump() # a JSON: json_str = user.model_dump_json() # desde JSON: user2 = User.model_validate_json(json_str)
model_dump() convierte a diccionario Python. model_dump_json() serializa a string JSON. model_validate_json() hace parse de JSON a modelo. Sustituyen a los antiguos .dict() y .json().
Validadores Custom
from pydantic import BaseModel, field_validator
class User(BaseModel):
email: str
@field_validator("email")
@classmethod
def validate_email(cls, v: str) -> str:
if "@" not in v:
raise ValueError("email inválido")
return v.lower()@field_validator crea validación personalizada. @classmethod es obligatorio. ValueError genera error 422. El valor retornado sustituye al original (normalización).
Model Config
from pydantic import BaseModel, ConfigDict
class User(BaseModel):
model_config = ConfigDict(
str_strip_whitespace=True,
str_to_lower=True,
frozen=True,
)
name: str
email: strConfigDict configura el comportamiento del modelo. str_strip_whitespace elimina espacios. frozen=True lo hace inmutable. str_to_lower normaliza a minúsculas.
Nested Models
class Address(BaseModel):
street: str
city: str
postal_code: str
class User(BaseModel):
name: str
addresses: list[Address] = []
primary: Address | None = NoneLos modelos pueden contener otros modelos. list[Address] valida arrays de objetos. Address | None acepta objeto o null. La validación es recursiva y automática.
Computed Fields
from pydantic import BaseModel, computed_field
class Rect(BaseModel):
width: float
height: float
@computed_field
@property
def area(self) -> float:
return self.width * self.height@computed_field añade campos calculados a la serialización. Aparece en el JSON de respuesta automáticamente. @property lo hace de solo lectura. Útil para totales y derivados.
Base de Dados
Setup SQLAlchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
SQLALCHEMY_URL = "sqlite:///./app.db"
engine = create_engine(SQLALCHEMY_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
passcreate_engine() crea la conexión a la BD. sessionmaker es una fábrica de sesiones. DeclarativeBase es la base para los modelos. check_same_thread es necesario para SQLite.
Relaciones
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
user: Mapped["User"] = relationship(back_populates="posts")
class User(Base):
posts: Mapped[list["Post"]] = relationship(back_populates="user")ForeignKey crea el enlace en la BD. relationship() define el acceso en Python. back_populates conecta ambos lados. Mapped[list] para relaciones one-to-many.
Pydantic + SQLAlchemy
from pydantic import BaseModel, ConfigDict
class UserOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
email: str
@app.get("/users/{id}", response_model=UserOut)
def get_user(id: int, db: Session = Depends(get_db)):
return db.query(User).filter_by(id=id).first()from_attributes=True permite crear Pydantic a partir de objetos ORM. Sustituye al antiguo orm_mode. El response_model filtra campos automáticamente. Puente entre BD y API.
Model SQLAlchemy
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import Mapped, mapped_column
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(unique=True)
active: Mapped[bool] = mapped_column(default=True)mapped_column() es la API moderna de SQLAlchemy 2.0. Mapped[type] define el tipo Python. primary_key marca la clave primaria. unique impide duplicados.
Migraciones (Alembic)
pip install alembic alembic init migrations # generar migración automática: alembic revision --autogenerate -m "crear users" # aplicar: alembic upgrade head # revertir: alembic downgrade -1
Alembic gestiona las migraciones de esquema. --autogenerate compara los modelos con la BD. upgrade head aplica todas las pendientes. downgrade -1 revierte la última. Esencial para la evolución del esquema.
Dependencia de Sesión
from sqlalchemy.orm import Session
from fastapi import Depends
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users")
def list_users(db: Session = Depends(get_db)):
return db.query(User).all()Depends(get_db) inyecta la sesión en cada request. yield mantiene la sesión abierta durante el endpoint. finally garantiza el cierre. Patrón de dependency injection de FastAPI.
Queries Avanzadas
from sqlalchemy import select, func # con select (SQLAlchemy 2.0): stmt = select(User).where(User.active == True).order_by(User.name) users = db.scalars(stmt).all() # agregaciones: total = db.scalar(select(func.count(User.id))) average = db.scalar(select(func.avg(User.age)))
select() es la API moderna de queries. scalars() retorna objetos directamente. func.count() y func.avg() hacen agregaciones. where() encadena condiciones.
Operaciones CRUD
# Crear user = User(name="Ana", email="ana@site.com") db.add(user) db.commit() db.refresh(user) # Leer users = db.query(User).filter(User.active == True).all() user = db.query(User).filter_by(id=1).first() # Eliminar db.delete(user) db.commit()
add() + commit() inserta en la BD. refresh() recarga con los datos generados (id). filter() usa expresiones, filter_by() usa kwargs. first() retorna None si no existe.
Async Database
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSession = async_sessionmaker(engine)
async def get_db():
async with AsyncSession() as session:
yield session
@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
return result.scalars().all()create_async_engine crea una engine asíncrona. async_sessionmaker es una fábrica de sesiones async. await db.execute() ejecuta queries sin bloquear. Requiere un driver async como asyncpg.
Autenticação e Segurança
Setup OAuth2 + JWT
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/profile")
def profile(token: str = Depends(oauth2_scheme)):
return {"token": token}OAuth2PasswordBearer extrae el token del header Authorization. tokenUrl indica el endpoint de login. El Swagger UI gana un botón "Authorize" automáticamente. Depends inyecta el token.
Endpoint de Login
from fastapi.security import OAuth2PasswordRequestForm
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
user = authenticate(form.username, form.password)
if not user:
raise HTTPException(401, "Credenciales inválidas")
token = create_token({"sub": str(user.id)})
return {"access_token": token, "token_type": "bearer"}OAuth2PasswordRequestForm espera username y password como form data. El response debe tener access_token y token_type. Compatible con el flujo OAuth2 del Swagger.
Generar Token JWT
from jose import jwt
from datetime import datetime, timedelta
SECRET_KEY = "super-secreto"
ALGORITHM = "HS256"
def create_token(data: dict, expires_min: int = 30):
payload = data.copy()
payload["exp"] = datetime.utcnow() + timedelta(minutes=expires_min)
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)jwt.encode() crea el token firmado. exp define la expiración (obligatorio). HS256 es el algoritmo de firma. SECRET_KEY debe estar en una variable de entorno.
Dependencias de Auth
from fastapi import Depends, HTTPException
def admin_required(user=Depends(get_current_user)):
if user.role != "admin":
raise HTTPException(403, "Acceso denegado")
return user
@app.delete("/users/{id}", dependencies=[Depends(admin_required)])
def delete_user(id: int): ...Las dependencias se encadenan: admin_required depende de get_current_user. dependencies=[] en el decorador protege sin inyectar. Retorna 403 para permisos insuficientes. Patrón reutilizable.
Validar Token
from jose import JWTError, jwt
from fastapi import HTTPException
def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401)
except JWTError:
raise HTTPException(status_code=401, detail="Token inválido")
return user_idjwt.decode() valida firma y expiración. sub es el subject (ID del user). JWTError captura tokens inválidos o expirados. Retorna 401 para credenciales inválidas.
CORS
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)CORSMiddleware permite requests de otros dominios. allow_origins lista los dominios permitidos. allow_credentials permite cookies. En producción, nunca uses ["*"] con credentials.
Hash de Password
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(password: str, hashed: str) -> bool:
return pwd_context.verify(password, hashed)CryptContext gestiona el hashing con salt automático. bcrypt es el esquema recomendado. hash() genera un hash irreversible. verify() compara el input con el hash guardado. Nunca guardes passwords en texto plano.
Rate Limiting
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@app.get("/api")
@limiter.limit("10/minute")
def api(request: Request):
return {"data": "..."}slowapi añade rate limiting. key_func identifica al cliente (por IP). limit() define el máximo por período. Retorna 429 cuando se excede. Esencial para APIs públicas.
Async e Background
Endpoint Async
@app.get("/data")
async def get_data():
result = await fetch_external()
return resultasync def define un endpoint asíncrono. await espera operaciones I/O sin bloquear. El servidor atiende otros requests mientras espera. Úsalo para BD, llamadas HTTP y archivos.
Lifespan Events
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# startup: abrir conexiones, cargar cache
print("Iniciando...")
yield
# shutdown: cerrar conexiones, limpiar recursos
print("Cerrando...")
app = FastAPI(lifespan=lifespan)lifespan sustituye a los antiguos eventos on_startup/on_shutdown. El código antes del yield corre en el arranque. El código después corre en el cierre. Ideal para inicializar recursos.
async vs sync
# async: para I/O (BD, HTTP, archivos)
@app.get("/a")
async def endpoint_a():
data = await external_api()
return data
# sync: para CPU (cálculos, procesamiento)
@app.get("/b")
def endpoint_b():
result = heavy_computation()
return resultUsa async para operaciones I/O (red, BD). Usa def normal para CPU-bound. Los endpoints sync corren en un threadpool automáticamente. Mezclar ambos es perfectamente válido.
Concurrencia con asyncio
import asyncio
@app.get("/parallel")
async def parallel():
results = await asyncio.gather(
fetch_api_1(),
fetch_api_2(),
fetch_api_3(),
)
return {"results": results}asyncio.gather() ejecuta múltiples tasks en paralelo. Todas corren concurrentemente en el event loop. Más rápido que llamadas secuenciales. Ideal para agregar datos de múltiples APIs.
Background Tasks
from fastapi import BackgroundTasks
def send_email(to: str, msg: str):
# tarea lenta (SMTP)
...
@app.post("/register")
def register(bg: BackgroundTasks):
bg.add_task(send_email, "ana@site.com", "¡Bienvenida!")
return {"msg": "Registrado"}BackgroundTasks se ejecuta tras enviar la respuesta. add_task() programa la función con argumentos. El cliente no espera a que la tarea termine. Ideal para emails, notificaciones y logs.
Streaming Response
from fastapi.responses import StreamingResponse
async def generate_data():
for i in range(1000):
yield f"línea {i}\n"
@app.get("/stream")
def stream():
return StreamingResponse(
generate_data(),
media_type="text/plain",
)StreamingResponse envía datos progresivamente. El generador yield produce chunks. media_type define el Content-Type. Ideal para archivos grandes, SSE y downloads.
httpx (HTTP async)
import httpx
@app.get("/external")
async def external():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()httpx.AsyncClient hace peticiones HTTP asíncronas. async with gestiona la conexión automáticamente. await client.get() no bloquea el event loop. Sustituto moderno de requests para async.
WebSockets
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket(ws: WebSocket):
await ws.accept()
while True:
data = await ws.receive_text()
await ws.send_text(f"Echo: {data}")@app.websocket crea un endpoint WebSocket. accept() establece la conexión. receive_text() y send_text() intercambian mensajes. Conexión persistente para tiempo real (chat, notificaciones).
Avançado e Deploy
Middleware
import time
@app.middleware("http")
async def request_time(request, call_next):
start = time.time()
response = await call_next(request)
duration = time.time() - start
response.headers["X-Process-Time"] = str(duration)
return response@app.middleware("http") intercepta todos los requests. call_next() pasa al siguiente handler. Permite añadir headers, logging o métricas. Se ejecuta antes y después del endpoint.
Tests con Override
from fastapi.testclient import TestClient
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)dependency_overrides sustituye dependencias en los tests. Cambia la BD real por una de prueba. El diccionario mapea original → sustituto. Limpia con .clear() tras los tests.
Buenas Prácticas
# 1. Type hints en todas partes # 2. Pydantic para validación (no manual) # 3. Dependencies para lógica compartida # 4. APIRouter para organizar # 5. response_model para controlar el output
Usa type hints siempre — generan validación gratis. Prefiere Pydantic a la validación manual. Depends() elimina la repetición. APIRouter escala el proyecto. response_model protege datos sensibles.
Manejo de Errores
from fastapi import HTTPException
@app.get("/items/{id}")
def get_item(id: int):
item = find_item(id)
if not item:
raise HTTPException(status_code=404, detail="Item no encontrado")
return item
# handler custom:
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
return JSONResponse(status_code=400, content={"error": str(exc)})HTTPException lanza errores HTTP con detalle. exception_handler captura excepciones custom globalmente. detail aparece en la respuesta JSON. Usa siempre status codes apropiados.
Deploy con Uvicorn
# Producción: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 # con Gunicorn: gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker # variables: # WEB_CONCURRENCY=4 # PORT=8000
--workers 4 crea múltiples procesos. Gunicorn como process manager en producción. UvicornWorker combina Gunicorn con ASGI. Regla: workers = 2 × CPU cores + 1.
Dependencias Reutilizables
from fastapi import Depends
class Pagination:
def __init__(self, page: int = 1, size: int = 10):
self.offset = (page - 1) * size
self.limit = size
@app.get("/items")
def list_items(pg: Pagination = Depends()):
return db.query(Item).offset(pg.offset).limit(pg.limit).all()Las clases con __init__ funcionan como dependencias. Depends() sin argumentos usa los defaults. Los parámetros se convierten en query params automáticamente. Reutilízalas en múltiples endpoints.
Docker
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"]
python:3.12-slim es la imagen base ligera. --no-cache-dir reduce el tamaño. EXPOSE documenta el puerto. --host 0.0.0.0 acepta conexiones externas. Multi-stage build para optimizar.
Tests con TestClient
from fastapi.testclient import TestClient
client = TestClient(app)
def test_root():
r = client.get("/")
assert r.status_code == 200
assert r.json() == {"msg": "¡Hola!"}
def test_create_item():
r = client.post("/items", json={"name": "Test", "price": 9.99})
assert r.status_code == 201TestClient simula requests HTTP sin servidor. json= envía body JSON. assert valida status y respuesta. Compatible con pytest. Ejecuta con pytest tests/.
Logging Estructurado
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@app.get("/items")
def list_items():
logger.info("Listado de ítems", extra={"user": "ana"})
return itemsEl logging de la stdlib es la forma estándar. extra añade contexto estructurado. level=INFO filtra por severidad. En producción usa JSON logging para agregación (ELK, Grafana).