Volver al inicio

Autorización JWT FastAPI: código paso a paso

Guía para implementar autenticación JWT en FastAPI para desarrolladores intermedios/seniors. Cubre estructura de token, esquemas Pydantic, módulo de seguridad, dependencias y endpoints protegidos con código completo.

JWT en FastAPI: código de login + rutas protegidas
Advertisement 728x90

Implementación de Autenticación JWT en FastAPI: Guía Completa para Desarrolladores

JWT (JSON Web Token) es un token compacto para transmitir información entre partes. El token consta de tres partes codificadas en Base64: encabezado, carga útil y firma, separadas por puntos.

Encabezado contiene el algoritmo de firma (por ejemplo, HS256) y el tipo de token (JWT):

{"alg": "HS256", "typ": "JWT"}

Carga útil almacena la información: sub (identificador del usuario), iat (momento de emisión), exp (tiempo de expiración):

Google AdInline article slot
{"sub": "alex", "name": "Alex Tomchok", "iat": 1516239022, "exp": 1516242622}

Firma se calcula como HMAC-SHA256 de (encabezado.carga_útil + clave_secreta). El servidor verifica la firma durante la validación: cualquier cambio en la carga útil invalida el token. Los datos son legibles pero están protegidos contra modificaciones.

Los tokens de acceso tienen una vida útil corta (15–30 minutos), mientras que los tokens de actualización son de larga duración (días/meses). Este ejemplo utiliza solo tokens de acceso.

Configuración del Proyecto y Dependencias

Crea la estructura:

Google AdInline article slot
fastapi-jwt-auth/
├── app/
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── schemas/
│   │   ├── token.py
│   │   └── user.py
│   ├── api/
│   │   └── v1/
│   │       ├── endpoints/
│   │       │   ├── auth.py
│   │       │   └── users.py
│   │       └── dependencies.py
│   └── services/
│       └── user_service.py
└── requirements.txt

requirements.txt:

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
pydantic[email]==2.5.0

Instala: pip install -r requirements.txt.

Configuración y Esquemas Pydantic

En core/config.py, define la configuración usando Pydantic:

Google AdInline article slot
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "tu-clave-super-secreta-cambia-esto-en-produccion"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

Esquemas en schemas/user.py:

from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool = True

    class Config:
        from_attributes = True

class UserInDB(User):
    hashed_password: str

schemas/token.py:

from pydantic import BaseModel

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

class TokenData(BaseModel):
    username: str | None = None

Seguridad: Hashing y Operaciones JWT

En core/security.py, implementa el hashing de contraseñas (bcrypt) y las funciones JWT:

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

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

Servicio de Usuario y Base de Datos Falsa

services/user_service.py con almacenamiento en memoria:

from app.schemas.user import UserInDB
from app.core.security import verify_password, get_password_hash

fake_users_db = {
    "alex": {
        "id": 1,
        "username": "alex",
        "email": "[email protected]",
        "hashed_password": get_password_hash("secret123"),
        "is_active": True,
    }
}

def get_user_by_username(username: str) -> UserInDB | None:
    if username in fake_users_db:
        return UserInDB(**fake_users_db[username])
    return None

def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user_by_username(username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

Dependencias para Autorización

En api/v1/dependencies.py, usa OAuth2PasswordBearer:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.core.security import decode_access_token
from app.services.user_service import get_user_by_username
from app.schemas.token import TokenData
from app.schemas.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="No se pudieron validar las credenciales",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    user = get_user_by_username(username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Usuario inactivo")
    return current_user

Endpoints: Inicio de Sesión y Rutas Protegidas

api/v1/endpoints/auth.py:

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.schemas.token import Token
from app.services.user_service import authenticate_user
from app.core.security import create_access_token

router = APIRouter(prefix="/auth", tags=["autenticación"])

@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Nombre de usuario o contraseña incorrectos",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token, token_type="bearer")

api/v1/endpoints/users.py:

from fastapi import APIRouter, Depends
from app.schemas.user import User
from app.api.v1.dependencies import get_current_active_user

router = APIRouter(prefix="/users", tags=["usuarios"])

@router.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

Conecta los routers en main.py y ejecuta uvicorn app.main:app.

Puntos Clave

  • La carga útil de JWT no está encriptada, solo firmada — usa solo datos no sensibles.
  • Almacena SECRET_KEY en variables de entorno, genera una con 32+ caracteres.
  • En producción, añade tokens de actualización y una base de datos (SQLAlchemy + Alembic).
  • Las dependencias de FastAPI permiten reutilizar la lógica de autorización sin duplicación.
  • Prueba los endpoints mediante Swagger UI: /docs.

— Editorial Team

Advertisement 728x90

Leer después