Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis

💻 Exemplo 1 — Emissão e Decodificação de JWT com Claims Padrão (sub, exp, iat)

from datetime import datetime, timedelta, timezone

import jwt

SECRET_KEY = "chave-secreta-de-no-minimo-32-bytes-em-producao"
ALGORITHM = "HS256"


def criar_access_token(dados: dict, expira_em: timedelta = timedelta(minutes=15)) -> str:
    payload = dados.copy()
    agora = datetime.now(timezone.utc)
    payload.update({"iat": agora, "exp": agora + expira_em})
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def decodificar_token(token: str) -> dict:
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])


token = criar_access_token({"sub": "user_101", "role": "EDITOR"})
print("Token gerado:", token[:40], "...")

payload = decodificar_token(token)
print("Payload decodificado:", payload)

try:
    token_vencido = criar_access_token({"sub": "user_101"}, expira_em=timedelta(seconds=-1))
    decodificar_token(token_vencido)
except jwt.ExpiredSignatureError:
    print("Token expirado corretamente rejeitado.")

💻 Exemplo 2 — Rota de Login com OAuth2PasswordRequestForm e Proteção com OAuth2PasswordBearer

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from pydantic import BaseModel

app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

BASE_USUARIOS = {
    "admin@empresa.com": {
        "email": "admin@empresa.com",
        "senha_hash": pwd_context.hash("admin123"),
    }
}


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


@app.post("/auth/token", response_model=TokenResponse)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    usuario = BASE_USUARIOS.get(form_data.username)
    if not usuario or not pwd_context.verify(form_data.password, usuario["senha_hash"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Usuário ou senha incorretos",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = criar_access_token({"sub": usuario["email"]})
    return TokenResponse(access_token=token)


async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
    credenciais_invalidas = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Não foi possível validar as credenciais",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decodificar_token(token)
    except jwt.PyJWTError:
        raise credenciais_invalidas

    email = payload.get("sub")
    usuario = BASE_USUARIOS.get(email)
    if usuario is None:
        raise credenciais_invalidas
    return usuario


@app.get("/perfil")
async def ler_perfil(usuario_atual: dict = Depends(get_current_user)):
    return {"email": usuario_atual["email"]}

💻 Exemplo 3 — Par Access/Refresh Token com Blacklist de Revogação

from datetime import timedelta

import jwt

ACCESS_TOKEN_EXPIRE = timedelta(minutes=15)
REFRESH_TOKEN_EXPIRE = timedelta(days=7)

# Em produção isto seria uma chave no Redis com TTL igual à validade
# restante do token — aqui, um set em memória para fins didáticos.
blacklist_refresh_tokens: set[str] = set()


def gerar_par_de_tokens(usuario_id: str) -> dict:
    access_token = criar_access_token({"sub": usuario_id, "type": "access"}, ACCESS_TOKEN_EXPIRE)
    refresh_token = criar_access_token({"sub": usuario_id, "type": "refresh"}, REFRESH_TOKEN_EXPIRE)
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"}


def renovar_access_token(refresh_token: str) -> dict:
    if refresh_token in blacklist_refresh_tokens:
        raise PermissionError("Refresh token revogado")

    payload = decodificar_token(refresh_token)
    if payload.get("type") != "refresh":
        raise PermissionError("Token informado não é um refresh token")

    novo_access_token = criar_access_token({"sub": payload["sub"], "type": "access"}, ACCESS_TOKEN_EXPIRE)
    return {"access_token": novo_access_token, "token_type": "bearer"}


def logout(refresh_token: str) -> None:
    blacklist_refresh_tokens.add(refresh_token)


par = gerar_par_de_tokens("user_101")
print("Par emitido:", {k: v[:20] + "..." for k, v in par.items() if k != "token_type"})

renovado = renovar_access_token(par["refresh_token"])
print("Access token renovado:", renovado["access_token"][:20], "...")

logout(par["refresh_token"])
try:
    renovar_access_token(par["refresh_token"])
except PermissionError as erro:
    print("Bloqueado após logout:", erro)