Objetivo
Crear una API REST minimal y segura con FastAPI que implemente registro de usuarios, login con JWT (access token), un endpoint protegido y pruebas automatizadas. Incluye Dockerfile y explicaciones de por qué tomar estas decisiones.
Requisitos previos
- Python 3.9+
- pip
- Conocimientos básicos de HTTP y Python
Estructura de carpetas
fastapi-jwt-example/
├─ app/
│ ├─ __init__.py
│ ├─ main.py
│ ├─ models.py
│ ├─ schemas.py
│ ├─ database.py
│ ├─ crud.py
│ ├─ auth.py
│ └─ dependencies.py
├─ tests/
│ └─ test_auth.py
├─ Dockerfile
└─ requirements.txt
requirements.txt
fastapi
uvicorn[standard]
sqlmodel
passlib[bcrypt]
python-jose[cryptography]
pytest
httpx
1) app/database.py
from sqlmodel import create_engine, Session
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
def get_session():
with Session(engine) as session:
yield session
2) app/models.py
from typing import Optional
from sqlmodel import SQLModel, Field
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
username: str = Field(index=True, unique=True)
hashed_password: str
is_active: bool = True
3) app/schemas.py
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
password: str
class UserRead(BaseModel):
id: int
username: str
is_active: bool
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
4) app/auth.py
from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import jwt
from typing import Optional
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# En producción, usa una variable de entorno segura
SECRET_KEY = "your-secret-key-change-me"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(subject: str, expires_delta: Optional[timedelta] = None) -> str:
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode = {"sub": subject, "exp": expire}
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str):
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
5) app/crud.py
from sqlmodel import Session, select
from .models import User
from .auth import get_password_hash, verify_password
def get_user_by_username(session: Session, username: str):
statement = select(User).where(User.username == username)
return session.exec(statement).first()
def create_user(session: Session, username: str, password: str) -> User:
hashed = get_password_hash(password)
user = User(username=username, hashed_password=hashed)
session.add(user)
session.commit()
session.refresh(user)
return user
def authenticate_user(session: Session, username: str, password: str):
user = get_user_by_username(session, username)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
6) app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlmodel import Session
from .database import get_session
from . import auth, crud
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")
async def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = auth.decode_access_token(token)
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = crud.get_user_by_username(session, username)
if user is None:
raise credentials_exception
return user
7) app/main.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import SQLModel
from .database import engine, get_session
from . import models, schemas, crud, auth
from .dependencies import get_current_user
from sqlmodel import Session
app = FastAPI()
# Crear tablas (solo para ejemplo local; en producción usa Alembic)
@app.on_event("startup")
def on_startup():
SQLModel.metadata.create_all(engine)
@app.post('/register', response_model=schemas.UserRead)
def register(user: schemas.UserCreate, session: Session = Depends(get_session)):
existing = crud.get_user_by_username(session, user.username)
if existing:
raise HTTPException(status_code=400, detail="Username already registered")
created = crud.create_user(session, user.username, user.password)
return schemas.UserRead(id=created.id, username=created.username, is_active=created.is_active)
@app.post('/login', response_model=schemas.Token)
def login(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
user = crud.authenticate_user(session, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=401, detail="Incorrect username or password")
token = auth.create_access_token(subject=user.username)
return {"access_token": token, "token_type": "bearer"}
@app.get('/me', response_model=schemas.UserRead)
def read_me(current_user: models.User = Depends(get_current_user)):
return schemas.UserRead(id=current_user.id, username=current_user.username, is_active=current_user.is_active)
8) Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY ./app ./app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
9) tests/test_auth.py
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, create_engine, Session
from app.main import app
from app.database import engine
from app import models
client = TestClient(app)
# Tests básicos: registro, login y acceso protegido
def test_register_login_and_protected():
# Asegurarse de que la BD esté limpia en este entorno de pruebas
SQLModel.metadata.create_all(engine)
# Registro
resp = client.post('/register', json={"username": "alice", "password": "secret"})
assert resp.status_code == 200
data = resp.json()
assert data['username'] == 'alice'
# Login
resp = client.post('/login', data={"username": "alice", "password": "secret"})
assert resp.status_code == 200
token = resp.json()['access_token']
assert token
# Acceso a endpoint protegido
headers = {"Authorization": f"Bearer {token}"}
resp = client.get('/me', headers=headers)
assert resp.status_code == 200
assert resp.json()['username'] == 'alice'
Explicación técnica y motivos
- FastAPI: rendimiento, documentación automática (OpenAPI) y tipado que facilita la validación.
- SQLModel: API simple basada en SQLAlchemy + Pydantic; reduce boilerplate entre modelos y schemas.
- Passlib (bcrypt): hashing de contraseñas probado y seguro. Nunca almacenar contraseñas en claro.
- python-jose: manejo JWT sencillo. Firmamos tokens con HS256 y una clave secreta (en producción usar variable de entorno y rotación).
- OAuth2PasswordBearer: esquema estándar para inyectar token desde Authorization header y facilitar la extracción con dependencias.
- Dependencias: usar get_current_user como dependencia permite reutilizar la lógica de autorización en cualquier endpoint.
- Tests con TestClient: pruebas rápidas de integración dentro del stack de FastAPI.
Por qué esta separación de responsabilidades
- database.py: centraliza el engine y la sesión. Facilita sustituir por otra base de datos.
- models.py y schemas.py: separan representación persistente (models) de la representación pública (schemas) para evitar exponer campos sensibles como hashed_password.
- crud.py: lógica de acceso a datos; facilita pruebas unitarias y mantenimiento.
- auth.py: concéntrate en hashing y JWT; mantener esta lógica aislada evita duplicación y ayuda a auditar seguridad.
- dependencies.py: encapsula la validación del token; hace que rutas sean más legibles.
Cómo ejecutar
- Crear un entorno virtual e instalar dependencias: python -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt
- Iniciar la app: uvicorn app.main:app --reload
- Probar endpoints: /docs para la UI interactiva (Swagger), /redoc para documentación.
- Ejecutar tests: pytest -q
Consideraciones de seguridad y siguientes pasos
- No uses claves estáticas en el código: carga SECRET_KEY desde variables de entorno o un vault. Implementa rotación de claves si es posible.
- Tokens de acceso cortos y tokens de refresco: para mayor seguridad, implementa refresh tokens y revocación (almacenar tokens en DB o blacklist).
- Usar HTTPS en producción y cabeceras de seguridad (CSP, HSTS, etc.).
- Separar configuración por entornos (p. ej. Pydantic Settings, 12-factor config).
Consejo avanzado: integra Alembic para migraciones y añade soporte para refresh tokens con almacenamiento en una tabla (o Redis) para poder revocar sesiones. Advertencia: audita y limita la información devuelta al cliente (por ejemplo, nunca exponer hashed_password ni claims sensibles en el token).
¿Quieres comentar?
Inicia sesión con Telegram para participar en la conversación