APIs REST con FastAPI y Pydantic: autenticación JWT, dependencias y despliegue práctico

En este artículo verás un patrón práctico para construir una API REST en Python usando FastAPI y Pydantic: modelos, validación, autenticación JWT, dependencias (DB session) y pruebas básicas. El objetivo es un punto de partida sólido que puedas adaptar rápidamente a producción.

Requisitos

fastapi
uvicorn[standard]
sqlmodel
asyncpg
passlib[bcrypt]
python-jose[cryptography]
httpx
pytest pytest-asyncio

Estructura mínima del proyecto

app/
├─ main.py
├─ db.py
├─ models.py
├─ schemas.py
├─ auth.py
├─ routers/
│  ├─ auth.py
│  └─ users.py
tests/
Dockerfile
requirements.txt

db.py — motor async y dependencia de sesión

from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://postgres:pass@localhost:5432/mydb")

engine: AsyncEngine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

models.py — tablas con SQLModel

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, nullable=False, unique=True)
    email: Optional[str] = None
    hashed_password: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

schemas.py — Pydantic para entrada/salida

from pydantic import BaseModel, EmailStr
from typing import Optional

class UserCreate(BaseModel):
    username: str
    email: Optional[EmailStr]
    password: str

class UserRead(BaseModel):
    id: int
    username: str
    email: Optional[EmailStr]

    class Config:
        orm_mode = True

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: Optional[str] = None

auth.py — hashing y JWT

from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import jwt, JWTError
from typing import Optional
import os

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = os.getenv("SECRET_KEY", "change_me_in_production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise

routers/auth.py — registro y login

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from fastapi.security import OAuth2PasswordRequestForm
from app.schemas import UserCreate, UserRead, Token
from app.models import User
from app.db import get_session
from app.auth import get_password_hash, verify_password, create_access_token

router = APIRouter()

@router.post('/register', response_model=UserRead)
async def register(user_in: UserCreate, session: AsyncSession = Depends(get_session)):
    q = select(User).where(User.username == user_in.username)
    result = await session.exec(q)
    if result.first():
        raise HTTPException(status_code=400, detail='Username already exists')
    user = User(username=user_in.username, email=user_in.email, hashed_password=get_password_hash(user_in.password))
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

@router.post('/token', response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), session: AsyncSession = Depends(get_session)):
    q = select(User).where(User.username == form_data.username)
    result = await session.exec(q)
    user = result.first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Incorrect credentials')
    access_token = create_access_token({"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

routers/users.py — dependencia de usuario actual y endpoint protegido

from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select
from app.schemas import UserRead
from app.models import User
from app.db import get_session
from app.auth import decode_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
router = APIRouter()

async def get_current_user(token: str = Depends(oauth2_scheme), session: AsyncSession = Depends(get_session)) -> User:
    try:
        payload = decode_token(token)
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid token')
    except Exception:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid token')
    q = select(User).where(User.username == username)
    result = await session.exec(q)
    user = result.first()
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='User not found')
    return user

@router.get('/users/me', response_model=UserRead)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

@router.post('/users/{user_id}/send-welcome')
async def send_welcome(user_id: int, background_tasks: BackgroundTasks, session: AsyncSession = Depends(get_session)):
    # ejemplo de tarea en background
    async def send_email_simulation(uid: int):
        # aquí encajarías una llamada a un proveedor de emails asíncrono
        print(f"Enviando welcome email a user {uid}")
    background_tasks.add_task(send_email_simulation, user_id)
    return {"status": "queued"}

main.py — ensamblado de la app

from fastapi import FastAPI
from app.routers import auth as auth_router, users as users_router
from app.db import init_db

app = FastAPI(title="Mi API con FastAPI")

app.include_router(auth_router.router)
app.include_router(users_router.router)

@app.on_event("startup")
async def on_startup():
    await init_db()

Pruebas básicas (pytest + httpx)

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_register_and_login():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        r = await ac.post('/register', json={"username": "alice", "password": "secret", "email": "a@a.com"})
        assert r.status_code == 200
        r = await ac.post('/token', data={"username": "alice", "password": "secret"})
        assert r.status_code == 200
        data = r.json()
        assert 'access_token' in data

Dockerfile mínimo

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80", "--proxy-headers"]

Buenas prácticas y seguridad

• No uses SECRET_KEY por defecto en producción: gestiona secretos con un vault o variables de entorno seguras.
• Usa hashing (passlib bcrypt) y nunca guardes passwords en claro.
• Limita el tiempo de vida del access_token y apoya la renovación mediante refresh tokens seguros.
• Valida scopes/roles en endpoints críticos: el simple sub en el token no es suficiente para autorización fina.
• Habilita HTTPS y cabeceras de seguridad en tu proxy (CSP, HSTS, etc.)

Próximo paso práctico: añade rate limiting (ej. Redis + fastapi-limiter), auditoría para endpoints sensibles y rotación de claves JWT. Si vas a producción, audita tus dependencias y habilita pruebas de integración contra un contenedor de Postgres real para asegurar que tu sesión async funciona como esperas.