Objetivos de Aprendizaje
- Entender la inyección de dependencias
- Usar Depends() para compartir código
- Crear dependencias de base de datos
- Implementar autenticación con Depends
Depends() (20 min)
La inyección de dependencias permite compartir código entre rutas.
Ejemplo Básico
from fastapi import Depends, FastAPI
app = FastAPI()
# Función de dependencia
def parametros_comunes(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# Usar Depends en una ruta
@app.get("/items/")
async def read_items(params: dict = Depends(parametros_comunes)):
return params
# Múltiples rutas usan la misma dependencia
@app.get("/users/")
async def read_users(params: dict = Depends(parametros_comunes)):
return params
# curl http://localhost:8000/items/?skip=20&limit=5
# {"skip": 20, "limit": 5}
Dependencias Anidadas
from fastapi import Depends, FastAPI
app = FastAPI()
def obtener_query(q: str | None = None):
return q
def obtener_skip(skip: int = 0):
return skip
# Depende de dos dependencias
def parametros_completos(
q: str | None = Depends(obtener_query),
skip: int = Depends(obtener_skip)
):
return {"q": q, "skip": skip}
@app.get("/items/")
async def read_items(params: dict = Depends(parametros_completos)):
return params
Conexión a Base de Datos (20 min)
Simular Conexión a BD
from fastapi import Depends, FastAPI
from typing import Generator
app = FastAPI()
# Simular conexión a base de datos
class Database:
def __init__(self):
self.connected = False
def connect(self):
self.connected = True
print("Conectando a BD...")
return self
def disconnect(self):
self.connected = False
print("Desconectando de BD...")
def query(self, sql: str):
return [{"id": 1, "name": "Item 1"}]
# Dependencia que maneja la conexión
def get_db() -> Generator:
db = Database()
try:
yield db.connect()
finally:
db.disconnect()
@app.get("/items/")
async def get_items(db: Database = Depends(get_db)):
# db ya está conectado
items = db.query("SELECT * FROM items")
return items
# Al finalizar la petición, se desconecta automáticamente
Patrón yield: El código antes del yield se ejecuta antes, el código después (finally) se ejecuta al terminar.
Autenticación (20 min)
En APIs modernas, se suele usar JWT (JSON Web Token) para manejar la autenticación de forma segura y escalable.
Verificar Token
from fastapi import Depends, FastAPI, HTTPException, Header
from typing import Optional
app = FastAPI()
# Base de datos simulada de usuarios
usuarios_db = {
"token-juan": {"username": "juan", "rol": "admin"},
"token-ana": {"username": "ana", "rol": "user"}
}
def verificar_token(x_token: str = Header(...)):
"""Extrae y verifica el token del header."""
if x_token not in usuarios_db:
raise HTTPException(status_code=401, detail="Token inválido")
return usuarios_db[x_token]
def requerir_admin(usuario: dict = Depends(verificar_token)):
"""Verifica que el usuario sea admin."""
if usuario["rol"] != "admin":
raise HTTPException(status_code=403, detail="Requiere rol de admin")
return usuario
# Ruta protegida con autenticación
@app.get("/perfil")
async def obtener_perfil(usuario: dict = Depends(verificar_token)):
return {"usuario": usuario}
# Ruta protegida solo para admins
@app.delete("/usuarios/{user_id}")
async def eliminar_usuario(
user_id: int,
admin: dict = Depends(requerir_admin)
):
return {"message": f"Usuario {user_id} eliminado", "por": admin["username"]}
# Probar:
# curl -H "X-Token: token-juan" http://localhost:8000/perfil
# curl -H "X-Token: token-invalido" http://localhost:8000/perfil # 401
Esquema de Dependencias
# Obtener usuario actual (requiere token)
async def get_current_user(token: str = Header(...)):
# Verificar token...
return {"id": 1, "username": "juan"}
# Verificar permisos (requiere usuario)
async def check_permissions(
user: dict = Depends(get_current_user),
required_role: str = "admin"
):
if user.get("role") != required_role:
raise HTTPException(403, "Sin permisos")
return user
# Ruta con múltiples dependencias
@app.post("/proyectos")
async def crear_proyecto(
proyecto: ProyectoCreate,
user: dict = Depends(check_permissions) # Token + Permisos
):
# Crear proyecto...
return proyecto
Ejercicio: API Protegida (10 min)
Crea una API de tareas con autenticación.
from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel
from typing import Optional, List
app = FastAPI()
# Modelos
class TareaCreate(BaseModel):
titulo: str
descripcion: Optional[str] = None
class Tarea(TareaCreate):
id: int
usuario_id: int
# Usuarios simulados
USUARIOS = {
"token123": {"id": 1, "username": "juan"},
"token456": {"id": 2, "username": "ana"}
}
# Base de datos simulada
tareas_db = []
contador_tarea = 1
# Dependencias
def get_current_user(x_token: str = Header(...)):
if x_token not in USUARIOS:
raise HTTPException(401, "Token inválido")
return USUARIOS[x_token]
# Rutas protegidas
@app.post("/tareas", response_model=Tarea)
async def crear_tarea(
tarea: TareaCreate,
usuario: dict = Depends(get_current_user)
):
global contador_tarea
nueva = Tarea(
id=contador_tarea,
usuario_id=usuario["id"],
**tarea.model_dump()
)
tareas_db.append(nueva)
contador_tarea += 1
return nueva
@app.get("/tareas/mis-tareas", response_model=List[Tarea])
async def listar_mis_tareas(
usuario: dict = Depends(get_current_user)
):
# Filtrar solo tareas del usuario autenticado
return [t for t in tareas_db if t.usuario_id == usuario["id"]]
@app.delete("/tareas/{tarea_id}")
async def eliminar_tarea(
tarea_id: int,
usuario: dict = Depends(get_current_user)
):
for i, tarea in enumerate(tareas_db):
if tarea.id == tarea_id and tarea.usuario_id == usuario["id"]:
del tareas_db[i]
return {"message": "Tarea eliminada"}
raise HTTPException(404, "Tarea no encontrada o no es tuya")
Beneficio: Las dependencias evitan repetir código de autenticación/BD en cada ruta.
Resumen
- Depends(): Inyecta dependencias en rutas
- yield: Gestiona recursos (conectar/desconectar BD)
- Reutilización: Múltiples rutas usan la misma dependencia
- Autenticación: Verificar token en dependencia
- Jerarquía: Dependencias pueden depender de otras
Práctica: Agrega una dependencia que verifique que el usuario esté activo antes de permitir crear tareas.