API REST con FastAPI, autenticación JWT y pruebas (proyecto práctico)

python API REST con FastAPI, autenticación JWT y pruebas (proyecto práctico)

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

  1. Crear un entorno virtual e instalar dependencias: python -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt
  2. Iniciar la app: uvicorn app.main:app --reload
  3. Probar endpoints: /docs para la UI interactiva (Swagger), /redoc para documentación.
  4. 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).

Comentarios
¿Quieres comentar?

Inicia sesión con Telegram para participar en la conversación


Comentarios (0)

Aún no hay comentarios. ¡Sé el primero en comentar!

Iniciar Sesión