Sistema de usuarios — Registro, inicio de sesión, autenticación JWT
En este capítulo aprenderás el esquema de autenticación JWT de FastAPI y entenderás la diferencia con la autenticación por Session de Django/Flask.
Autenticación JWT vs Session
Django y Flask usanautenticación por Session: el servidor almacena el estado del usuario, y en la Cookie solo se guarda el Session ID.
FastAPI recomiendaautenticación JWT: después de iniciar sesión, el usuario obtiene un Token, y las solicitudes posteriores pueden autenticarse llevando ese Token.
| Característica | Session（Django/Flask） | JWT（FastAPI） |
|---|---|---|
| Ubicación de almacenamiento | Lado del servidor (memoria/base de datos) | Lado del cliente (Cookie/LocalStorage) |
| Escalabilidad | El servidor necesita compartir la Session | Sin estado, soporta distribución de forma natural |
| Escenarios aplicables | Sitios web renderizados en el servidor | API + SPA / Móvil |
| Control de expiración | El servidor puede invalidarlo en cualquier momento | El Token permanece válido mientras no expire |
Instalar dependencias
(venv) $ pip install passlib python-jose python-multipart
- passlib: hash de contraseñas (equivalente a werkzeug.security)
- python-jose: generación y verificación de JWT
- python-multipart: manejo de envío de formularios (formulario de inicio de sesión OAuth2)
Definir el modelo User
Ejemplo
# Archivo: models.py (nuevo)
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""Hash de contraseña"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""Verificar contraseña"""
return pwd_context.verify(password, self.hashed_password)
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""Hash de contraseña"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""Verificar contraseña"""
return pwd_context.verify(password, self.hashed_password)
Generar y ejecutar migraciones:
(venv) $ alembic revision --autogenerate -m "新增 User 模型" (venv) $ alembic upgrade head
Funciones de utilidad JWT
Ejemplo
# Archivo: auth.py (nuevo archivo)
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# Clave (en producción se lee desde variables de entorno)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Validez del Token: 24 horas
# OAuth2PasswordBearer le indica a FastAPI: extraer el JWT del encabezado de solicitud Authorization: Bearer <token>
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""Generar JWT Token"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
"""Verificar y decodificar JWT Token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# Clave (en producción se lee desde variables de entorno)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Validez del Token: 24 horas
# OAuth2PasswordBearer le indica a FastAPI: extraer el JWT del encabezado de solicitud Authorization: Bearer <token>
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""Generar JWT Token"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
"""Verificar y decodificar JWT Token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
Ruta de registro
Ejemplo
# Archivo: routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=[Usuario])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""Registro de usuario"""
# Comprobar si el nombre de usuario y el correo ya existen
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail=El nombre de usuario ya está en uso)
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail=El correo ya está registrado)
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=[Usuario])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""Registro de usuario"""
# Comprobar si el nombre de usuario y el correo ya existen
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail=El nombre de usuario ya está en uso)
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail=El correo ya está registrado)
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
Ruta de inicio de sesión (emitir JWT)
Ejemplo
# Archivo: routers/users.py (agregar)
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
Iniciar sesión y obtener Token
Usa OAuth2PasswordRequestForm: los nombres de campo son username y password (formato de formulario)
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail=Nombre de usuario o contraseña incorrectos,
headers={"WWW-Authenticate": "Bearer"}
)
# Emitir JWT: solo incluye user_id (no se deben poner datos sensibles)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
Iniciar sesión y obtener Token
Usa OAuth2PasswordRequestForm: los nombres de campo son username y password (formato de formulario)
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail=Nombre de usuario o contraseña incorrectos,
headers={"WWW-Authenticate": "Bearer"}
)
# Emitir JWT: solo incluye user_id (no se deben poner datos sensibles)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
En el JWT Token
subestá el campo estándar subject, que almacena el identificador del usuario. El Token en sí no está cifrado (solo está codificado en Base64), por lo tantonunca debesponer información sensible como contraseñas. Si quieres cifrarlo, usa JWE en lugar de JWT.
Usar protección JWT en las rutas
Ejemplo
# Función de dependencia para obtener el usuario actual
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Obtener el usuario actual desde el JWT Token"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail=Token inválido o expirado)
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail=El usuario no existe)
return user
# Inyectar en las rutas que requieren inicio de sesión
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""Obtener la información del usuario actual (requiere inicio de sesión)"""
return current_user
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Obtener el usuario actual desde el JWT Token"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail=Token inválido o expirado)
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail=El usuario no existe)
return user
# Inyectar en las rutas que requieren inicio de sesión
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""Obtener la información del usuario actual (requiere inicio de sesión)"""
return current_user
Para páginas SSR, el JWT generalmente se guarda en la Cookie (en lugar del encabezado Authorization), y en las rutas de plantillas se extrae el Token de la Cookie para verificar.
Resumen del capítulo
En este capítulo implementaste la autenticación JWT de FastAPI: passlib para el hash de contraseñas, python-jose para emitir y verificar JWT, OAuth2PasswordBearer para extraer el Token, y Depends para la inyección de dependencias de autenticación.
A diferencia de la autenticación por Session de Django/Flask, JWT es un esquema sin estado, más adecuado para escenarios con separación de frontend y backend.
Otras extensiones