Fuck Learn

Appearance

🎯 Meta: dar el salto del Python "clásico" (Flask) al Python moderno: tipado estático, validación automática, async/await y documentación interactiva generada sola. FastAPI es hoy el framework Python más deseado para APIs.

Versiones: Python 3.13 · FastAPI 0.136 · Pydantic 2 · Uvicorn.

5.1 · ¿Por qué FastAPI supera a Flask para APIs? ​

Tres superpoderes que Flask no tiene de fábrica:

  1. Validación automática por tipos: declaras los tipos con anotaciones de Python y FastAPI valida, convierte y documenta solo. Menos código, menos bugs.
  2. Documentación interactiva gratis: abre /docs y tienes una interfaz (Swagger UI) donde probar tu API sin Postman. Se genera sola desde tu código.
  3. Async nativo: maneja miles de conexiones concurrentes (ideal para I/O: BD, APIs externas).

🧠 La idea clave de FastAPI: "el código es la documentación y la validación". Un mismo tipo de dato (Pydantic model) sirve para validar la entrada, documentar y serializar la salida. Se acabó duplicar esfuerzo.

5.2 · Instalación y primer endpoint ​

bash
python -m venv .venv && source .venv/bin/activate   # (Windows: .venv\Scripts\Activate.ps1)
pip install "fastapi[standard]"                      # incluye uvicorn y utilidades
python
# main.py
from fastapi import FastAPI

app = FastAPI(title="Tienda API", version="1.0")

@app.get("/")
def home():
    return {"mensaje": "¡Hola, FastAPI!"}
bash
fastapi dev main.py          # servidor de desarrollo con recarga

Abre http://localhost:8000/docs → ya tienes documentación interactiva. Prueba el endpoint desde ahí. Eso es lo que engancha a todo el mundo con FastAPI.

5.3 · Pydantic — El corazón de FastAPI ​

Pydantic define la "forma" de tus datos con tipos. Un modelo Pydantic valida automáticamente:

python
# schemas.py
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

# Lo que llega al crear (entrada)
class ProductoCrear(BaseModel):
    nombre: str = Field(min_length=1, max_length=255)
    precio: float = Field(gt=0, description="Precio en soles, mayor que 0")
    stock: int = Field(default=0, ge=0)

# Lo que devolvemos (salida)
class ProductoSalida(BaseModel):
    id: int
    nombre: str
    precio: float
    disponible: bool
    creado_en: datetime

    model_config = {"from_attributes": True}   # permite crear desde objetos ORM

Si alguien manda precio: -5 o se olvida el nombre, FastAPI responde 422 con un error detallado automáticamente. Tú no escribes ninguna validación.

💡 Tip: separa siempre el modelo de entrada del de salida. Nunca aceptes un id o un campo es_admin desde el cliente (lo pondría a mano un atacante); nunca devuelvas la contraseña. Modelos distintos = seguridad por diseño.

5.4 · CRUD con validación y tipos ​

python
# main.py
from fastapi import FastAPI, HTTPException, status
from schemas import ProductoCrear, ProductoSalida

app = FastAPI(title="Tienda API")

# "BD" en memoria para el ejemplo (luego la cambiamos por PostgreSQL)
db: dict[int, dict] = {}
contador = 0

@app.get("/productos", response_model=list[ProductoSalida])
def listar():
    return list(db.values())

@app.get("/productos/{id}", response_model=ProductoSalida)
def ver(id: int):
    if id not in db:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Producto no encontrado")
    return db[id]

@app.post("/productos", response_model=ProductoSalida, status_code=201)
def crear(datos: ProductoCrear):        # ← validación automática por el tipo
    global contador
    contador += 1
    producto = {
        "id": contador, **datos.model_dump(),
        "disponible": datos.stock > 0,
        "creado_en": datetime.now(),
    }
    db[contador] = producto
    return producto

@app.put("/productos/{id}", response_model=ProductoSalida)
def editar(id: int, datos: ProductoCrear):
    if id not in db:
        raise HTTPException(404, "Producto no encontrado")
    db[id].update(datos.model_dump(), disponible=datos.stock > 0)
    return db[id]

@app.delete("/productos/{id}", status_code=204)
def borrar(id: int):
    if id not in db:
        raise HTTPException(404, "Producto no encontrado")
    del db[id]

Fíjate: el tipo de cada parámetro hace el trabajo. id: int valida que sea entero (y da 422 si mandas texto). datos: ProductoCrear valida todo el cuerpo. Es Python puro llevado al extremo.

5.5 · Base de datos real: SQLModel + PostgreSQL ​

Del creador de FastAPI existe SQLModel (combina SQLAlchemy + Pydantic en una sola clase):

bash
pip install sqlmodel "psycopg[binary]"
python
# models.py
from sqlmodel import SQLModel, Field
from datetime import datetime, timezone

class Producto(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    nombre: str = Field(max_length=255)
    precio: float
    stock: int = 0
    activo: bool = True
    creado_en: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
python
# database.py
from sqlmodel import create_engine, Session, SQLModel

URL = "postgresql+psycopg://postgres:secreto@localhost:5432/tienda"
engine = create_engine(URL, echo=True)          # echo=True muestra el SQL generado

def crear_tablas():
    SQLModel.metadata.create_all(engine)

def get_session():                              # dependencia de sesión
    with Session(engine) as session:
        yield session

Inyección de dependencias (Depends) ​

FastAPI tiene un sistema de inyección de dependencias elegantísimo. La sesión de BD se "inyecta" en cada endpoint:

python
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from database import engine, crear_tablas, get_session
from models import Producto

app = FastAPI()

@app.on_event("startup")
def inicio():
    crear_tablas()

@app.get("/productos")
def listar(session: Session = Depends(get_session)):
    return session.exec(select(Producto)).all()

@app.post("/productos", status_code=201)
def crear(producto: Producto, session: Session = Depends(get_session)):
    session.add(producto)
    session.commit()
    session.refresh(producto)      # recarga con el id generado
    return producto

@app.get("/productos/{id}")
def ver(id: int, session: Session = Depends(get_session)):
    producto = session.get(Producto, id)
    if not producto:
        raise HTTPException(404, "No encontrado")
    return producto

🧠 Inyección de dependencias (DI): en vez de que cada función cree su propia conexión a la BD, la pide como parámetro y FastAPI se la da. Esto facilita testear (puedes inyectar una BD falsa) y es el mismo concepto que verás llevado al extremo en NestJS (cap. 06). Es un pilar de SOLID (la "D": inversión de dependencias, cap. 10).

5.6 · Async/await — Concurrencia ​

Cuando tu endpoint espera algo externo (BD, otra API), async permite atender otras peticiones mientras tanto, en vez de quedarse bloqueado:

python
import httpx

@app.get("/clima/{ciudad}")
async def clima(ciudad: str):
    async with httpx.AsyncClient() as client:
        # mientras esperamos la respuesta, el servidor atiende otras peticiones
        r = await client.get(f"https://api.clima.com/v1/{ciudad}")
        return r.json()

🧠 ¿Cuándo usar async? Cuando esperas I/O (red, disco, BD). Para cálculo puro (CPU) no ayuda. Regla práctica: si la función hace await de algo, hazla async. Usa drivers async (asyncpg, httpx) para aprovecharlo de verdad.

5.6.1 · Lifespan — arranque y apagado ordenados ​

@app.on_event("startup") (usado en 5.5) está deprecado. La forma moderna es un gestor de contexto de lifespan: código antes del yield corre al arrancar, después del yield corre al apagar (cierre de conexiones, flush de métricas):

python
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    crear_tablas()                       # arranque: se ejecuta UNA vez
    pool = await crear_pool_conexiones()
    app.state.pool = pool                # estado compartido, accesible como request.app.state.pool
    yield
    await pool.close()                   # apagado: libera recursos al parar el servidor

app = FastAPI(lifespan=lifespan)

🧠 Por qué importa: sin un cierre ordenado, las conexiones de BD/Redis quedan abiertas cuando el proceso recibe SIGTERM (p. ej. al desplegar una nueva versión en Kubernetes, apéndice G) — el pool se agota poco a poco entre despliegues hasta que la app deja de poder conectar.

5.7 · Autenticación con JWT (OAuth2) ​

FastAPI trae utilidades de seguridad OAuth2 + JWT:

python
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
import jwt   # pip install pyjwt

oauth2 = OAuth2PasswordBearer(tokenUrl="login")
SECRET = "clave_secreta_del_env"

@app.post("/login")
def login(usuario: str, password: str):
    # ...verificar credenciales contra la BD...
    token = jwt.encode({"sub": usuario}, SECRET, algorithm="HS256")
    return {"access_token": token, "token_type": "bearer"}

def usuario_actual(token: str = Depends(oauth2)):
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
        return payload["sub"]
    except jwt.PyJWTError:
        raise HTTPException(401, "Token inválido")

@app.get("/perfil")
def perfil(usuario: str = Depends(usuario_actual)):   # ruta protegida
    return {"usuario": usuario}

⚠️ Guarda las contraseñas con hashing (bcrypt/argon2, vía passlib), nunca en texto plano. Y el SECRET en variables de entorno. La seguridad de JWT la profundizas en tu Manual de Ciberseguridad.

5.8 · Tests con pytest + TestClient ​

FastAPI incluye un cliente de pruebas basado en httpx:

python
# tests/test_productos.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_crear_producto():
    resp = client.post("/productos", json={"nombre": "Mouse", "precio": 49.9})
    assert resp.status_code == 201
    assert resp.json()["nombre"] == "Mouse"

def test_precio_negativo_falla():
    resp = client.post("/productos", json={"nombre": "X", "precio": -1})
    assert resp.status_code == 422    # Pydantic lo rechaza solo

def test_ver_inexistente():
    resp = client.get("/productos/9999")
    assert resp.status_code == 404

Para tests con BD, inyecta una sesión de test sobrescribiendo la dependencia:

python
app.dependency_overrides[get_session] = get_session_de_test

💡 Tip: dependency_overrides es la joya del testing en FastAPI. Como todo se inyecta con Depends, en los tests reemplazas la BD real por una de prueba con una línea. Esto es DI pagando dividendos.

bash
pytest -v

5.8.1 · BackgroundTasks y respuestas en streaming ​

Para trabajo rápido que no necesita ser una cola completa (apéndice B), BackgroundTasks lo ejecuta después de devolver la respuesta al cliente:

python
from fastapi import BackgroundTasks

def registrar_log(mensaje: str):
    with open("accesos.log", "a") as f:
        f.write(mensaje + "\n")

@app.post("/productos", status_code=201)
def crear(datos: ProductoCrear, tareas: BackgroundTasks):
    producto = guardar(datos)
    tareas.add_task(registrar_log, f"producto {producto['id']} creado")  # corre tras responder
    return producto

⚠️ BackgroundTasks corre en el mismo proceso que atendió la petición: si el worker muere o reinicia, la tarea pendiente se pierde sin reintento. Para algo que debe ejecutarse sí o sí (cobrar, enviar un email crítico), usa una cola de verdad (Celery/RQ, o Redis Streams del apéndice B) — BackgroundTasks es para trabajo "best effort" (logging, analítica no crítica).

Streaming — para respuestas grandes o generadas incrementalmente (el mismo patrón que verás con LLMs en el apéndice O):

python
from fastapi.responses import StreamingResponse

def generar_csv(filas: list[dict]):
    yield "id,nombre,precio\n"
    for f in filas:
        yield f"{f['id']},{f['nombre']},{f['precio']}\n"     # se envía trozo a trozo

@app.get("/productos/exportar")
def exportar():
    return StreamingResponse(generar_csv(list(db.values())), media_type="text/csv")

💡 El cliente empieza a recibir bytes antes de que termine de generarse todo el contenido — crucial para exportaciones grandes (no esperas a construir el CSV entero en memoria) y para streaming token a token de un LLM (apéndice O).

5.9 · Producción (Uvicorn + Gunicorn) ​

bash
# desarrollo:
fastapi dev main.py

# producción: Uvicorn con varios workers gestionados por Gunicorn
pip install gunicorn uvicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

Delante, Nginx (cap. 13). En Docker (cap. 12), la imagen python:3.13-slim es tu base.

5.10 · Buenas prácticas FastAPI ​

  1. Modelos de entrada ≠ de salida. Nunca expongas ni aceptes campos sensibles.
  2. response_model siempre. Documenta y filtra la salida automáticamente.
  3. Todo por Depends (BD, usuario actual, config): testeable y limpio.
  4. async para I/O, con drivers async (asyncpg). No mezcles llamadas bloqueantes en funciones async.
  5. Routers por módulo (APIRouter) para organizar, como los blueprints de Flask.
  6. Migraciones con Alembic (SQLModel no las gestiona solo).
  7. /docs es tu amigo, pero ciérralo o protégelo en producción si la API es privada.

Organización con routers (equivalente a blueprints):

python
# routers/productos.py
from fastapi import APIRouter
router = APIRouter(prefix="/productos", tags=["productos"])

@router.get("/")
def listar(): ...

# main.py
from routers import productos
app.include_router(productos.router)

✅ Ejercicio del capítulo ​

La API del blog, ahora en FastAPI, aprovechando sus superpoderes:

1. SQLModel: Articulo y Comentario con relación.
2. Modelos Pydantic separados de entrada/salida.
3. CRUD con inyección de sesión por Depends.
4. Login con JWT y proteger crear/editar/borrar.
5. Un endpoint async que consuma una API externa (ej. una imagen aleatoria).
6. Tests con TestClient + dependency_overrides (mínimo 5).
7. Explora /docs y prueba todo desde ahí.
8. Migra el `on_event("startup")` a un `lifespan` con arranque y apagado.
9. Añade un `BackgroundTasks` que registre un log al crear un artículo.

Compara los 3 mundos que ya conoces: Laravel (todo hecho, PHP), Flask (minimalista, montas tú), FastAPI (tipado y automático). Ya piensas en conceptos, no en sintaxis. Ahora subimos a TypeScript.

💡 Pistas de la solución
  • La relación Articulo↔Comentario en SQLModel: comentarios: list["Comentario"] = Relationship(back_populates="articulo") en Articulo, y articulo_id: int = Field(foreign_key="articulo.id") en Comentario.
  • Para el endpoint async con API externa, usa httpx.AsyncClient (5.6) — si usas requests (bloqueante) dentro de una función async def, bloqueas el event loop entero: peor que no usar async.
  • El fallo típico del punto 6: sobrescribir dependency_overrides DESPUÉS de crear el TestClient no sirve — hazlo antes, o usa un fixture de pytest con yield que lo limpie tras cada test (app.dependency_overrides.clear()).

🧠 Autoevaluación ​

Porque son responsabilidades distintas: el de entrada define qué puede mandar el cliente (y evita que alguien inyecte campos como es_admin o id); el de salida define qué se expone (y evita filtrar contraseñas u otros campos internos). Usar el mismo modelo para ambos es aceptar por accidente lo que el cliente decida mandar.

Testeabilidad: como la dependencia se "pide" en vez de crearse dentro, puedes sustituirla por una versión de prueba con dependency_overrides sin tocar el código del endpoint. Es inversión de dependencias (la "D" de SOLID, cap. 10) aplicada de forma nativa por el framework.

Bloquea el event loop completo: mientras esa llamada bloqueante se ejecuta, el servidor no puede atender NINGUNA otra petición, ni siquiera las que no dependen de esa librería. Es peor que usar una función síncrona normal, porque rompe la promesa de concurrencia que async hace al resto de la app.

Porque filtra y valida la salida automáticamente incluso si el objeto que devuelves tiene campos de más (por ejemplo, un modelo de BD con una contraseña hasheada) — response_model garantiza que solo salga lo declarado, no lo que "por ahora" devuelve tu función.

Porque lifespan es un único gestor de contexto con arranque (antes del yield) y apagado (después del yield) explícitos y ordenados, mientras que los eventos separados de on_event no garantizan el mismo control sobre cuándo se liberan los recursos. Sin un cierre ordenado, un pool de conexiones puede agotarse progresivamente entre despliegues.

Porque corre en el mismo proceso que respondió la petición: si el worker muere o reinicia antes de completar la tarea, se pierde sin ningún reintento. Sirve para trabajo "best effort" (logging, analítica no crítica); para algo que debe ejecutarse sí o sí, hace falta una cola persistente con reintentos.

Siguiente: 06-node-nestjs.md — arquitectura empresarial con TypeScript.