📚 Módulo 05: Backend - API REST, CORS e Segurança com JWT

Neste módulo prático do backend, exporemos nossa lógica de serviços para a internet! Criaremos controladores REST assíncronos usando APIRouters, habilitaremos a segurança de origem com o CORS Middleware para conectar a API com o Vue 3 e construiremos o mecanismo de autenticação stateless JWT utilizando a injeção de dependências do FastAPI.

---

🔒 1. Segurança com JWT no FastAPI

No ecossistema FastAPI, o fluxo de segurança stateless é incrivelmente elegante. Utiliza-se a injeção de dependência (Depends) para validar e decodificar tokens Bearer, extrair o usuário e verificar seus privilégios de acesso (Admin ou Cliente) antes do endpoint rodar.

1. Injeção de Sessão do Banco (get_db)

Primeiro, criamos a dependência que gerencia o ciclo de vida das conexões assíncronas do banco, fechando-as de forma limpa ao final de cada requisição.

Crie o arquivo em app/database.py:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

# Injetado nos routers para expor a transação limpa
async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

2. Utilitário JWT e Dependência de Segurança (app/security.py)

Aqui criaremos o gerador de tokens e o validador que interceptará as chamadas protegidas da API.

sequenceDiagram
    participant Front as Vue SPA
    participant Router as Auth Router
    participant Sec as Security (jwt)
    participant DB as Banco de Dados

    Front->>Router: POST /login (user, pass)
    Router->>DB: Busca Usuário
    DB-->>Router: Hash bcrypt
    Router->>Sec: Valida Hash
    Sec-->>Router: Hash Válido
    Router->>Sec: criar_token_acesso()
    Sec-->>Router: eyJhbGci...
    Router-->>Front: { token: "eyJ..." }

    Note over Front,DB: Próximas requisições protegidas
    Front->>Sec: GET /api/pedidos (Header: Bearer eyJ...)
    Sec->>Sec: jwt.decode(token)
    Sec-->>Front: Permite o acesso

Crie o arquivo app/security.py:

from datetime import datetime, timedelta
from typing import List
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from passlib.context import CryptContext
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database import get_db
from app.models.usuario import Usuario

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")

# Cria token Bearer expirável
def criar_token_acesso(data: dict) -> str:
    dados_token = data.copy()
    expira = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    dados_token.update({"exp": expira})
    return jwt.encode(dados_token, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)

# Injetado nos endpoints protegidos para ler e validar o JWT
async def obter_usuario_logado(
    token: str = Depends(oauth2_scheme), 
    db: AsyncSession = Depends(get_db)
) -> Usuario:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Credenciais inválidas ou token expirado.",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    query = select(Usuario).where(Usuario.username == username)
    result = await db.execute(query)
    usuario = result.scalar_one_or_none()
    if usuario is None:
        raise credentials_exception
    return usuario

# Filtro de privilégio para rotas Admin
class RoleRequired:
    def __init__(self, papeis_permitidos: List[str]):
        self.papeis_permitidos = papeis_permitidos

    def __call__(self, usuario: Usuario = Depends(obter_usuario_logado)):
        if usuario.papel not in self.papeis_permitidos:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Acesso negado. Nível de privilégio insuficiente."
            )
        return usuario

---

🚪 2. Roteador de Autenticação (app/routers/auth_router.py)

Cria o endpoint /api/auth/login que recebe as credenciais em JSON, verifica o hash do banco via bcrypt e retorna o token de acesso Bearer JWT.

Crie o arquivo em app/routers/auth_router.py:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, EmailStr
from app.database import get_db
from app.models.usuario import Usuario
from app.security import pwd_context, criar_token_acesso

router = APIRouter(prefix="/api/auth", tags=["Autenticação"])

class LoginRequest(BaseModel):
    username: EmailStr
    password: str

class LoginResponse(BaseModel):
    token: str
    username: str
    papel: str

@router.post("/login", response_model=LoginResponse)
async def login(req: LoginRequest, db: AsyncSession = Depends(get_db)):
    query = select(Usuario).where(Usuario.username == req.username)
    result = await db.execute(query)
    usuario = result.scalar_one_or_none()

    if not usuario or not pwd_context.verify(req.password, usuario.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, 
            detail="Usuário ou senha incorretos."
        )

    token = criar_token_acesso({"sub": usuario.username})
    
    return LoginResponse(
        token=token,
        username=usuario.username,
        papel=usuario.papel
    )

---

💻 3. Roteadores REST de Produtos e Pedidos

1. Roteador de Produtos (app/routers/produto_router.py)

Exibe a vitrine pública e protege as escritas (POST, PUT, DELETE) com o privilégio ROLE_ADMIN.

Crie o arquivo em app/routers/produto_router.py:

from typing import List
from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.produto_schema import ProdutoFormSchema, ProdutoSchema
from app.services.produto_service import ProdutoService
from app.security import RoleRequired

router = APIRouter(prefix="/api/produtos", tags=["Produtos"])

# Rota pública: Catálogo da loja
@router.get("", response_model=List[ProdutoSchema])
async def listar(db: AsyncSession = Depends(get_db)):
    service = ProdutoService(db)
    return await service.listar_todos()

# Rota pública: Detalhe do eletrônico
@router.get("/{id}", response_model=ProdutoSchema)
async def obter_por_id(id: int, db: AsyncSession = Depends(get_db)):
    service = ProdutoService(db)
    return await service.buscar_por_id(id)

# Rota pública: Pesquisa dinâmica
@router.get("/pesquisa/", response_model=List[ProdutoSchema])
async def pesquisar(nome: str, db: AsyncSession = Depends(get_db)):
    service = ProdutoService(db)
    return await service.pesquisar_por_nome(nome)

# Protegida: Somente administradores cadastram
@router.post("", response_model=ProdutoSchema, status_code=status.HTTP_201_CREATED)
async def criar(
    form: ProdutoFormSchema, 
    db: AsyncSession = Depends(get_db),
    _ = Depends(RoleRequired(["ROLE_ADMIN"]))
):
    service = ProdutoService(db)
    return await service.criar(form)

# Protegida: Somente administradores editam
@router.put("/{id}", response_model=ProdutoSchema)
async def atualizar(
    id: int, 
    form: ProdutoFormSchema, 
    db: AsyncSession = Depends(get_db),
    _ = Depends(RoleRequired(["ROLE_ADMIN"]))
):
    service = ProdutoService(db)
    return await service.atualizar(id, form)

# Protegida: Somente administradores apagam
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def deletar(
    id: int, 
    db: AsyncSession = Depends(get_db),
    _ = Depends(RoleRequired(["ROLE_ADMIN"]))
):
    service = ProdutoService(db)
    await service.deletar(id)

2. Roteador de Pedidos (app/routers/pedido_router.py)

Protegido para checkout, exigindo login (obter_usuario_logado).

Crie o arquivo em app/routers/pedido_router.py:

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.pedido_schema import PedidoFormSchema, PedidoSchema
from app.services.pedido_service import PedidoService
from app.security import obter_usuario_logado

router = APIRouter(prefix="/api/pedidos", tags=["Pedidos"])

@router.post("", response_model=PedidoSchema, status_code=status.HTTP_201_CREATED)
async def checkout(
    form: PedidoFormSchema, 
    db: AsyncSession = Depends(get_db),
    _ = Depends(obter_usuario_logado)
):
    service = PedidoService(db)
    return await service.realizar_pedido(form)

---

🌐 4. Ativação do CORS e Inicialização da API (app/main.py)

Para que nosso frontend Vue 3 consiga consumir a API sem ser bloqueado pela segurança do navegador, ativamos o CORSMiddleware apontando para as URLs de desenvolvimento e produção da Netlify.

Edite o arquivo central app/main.py:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import auth_router, produto_router, pedido_router

app = FastAPI(title="TecLoja 02 API", version="1.0.0")

# Lista de domínios permitidos (CORS Origin whitelist)
origins = [
    "http://localhost:5173",       # Porta padrão do Vite local
    "http://localhost:3000",
    "https://tecloja-vue.netlify.app" # Link de produção da SPA Netlify
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"], # Permite todos os verbos (GET, POST, PUT, DELETE, OPTIONS)
    allow_headers=["*"], # Permite todos os cabeçalhos (incluindo Authorization)
)

# Acoplamento dos Roteadores REST
app.include_router(auth_router.router)
app.include_router(produto_router.router)
app.include_router(pedido_router.router)

@app.get("/")
async def root():
    return {"status": "TecLoja 02 API rodando perfeitamente de forma assíncrona!"}

Teste e suba a API localmente rodando em seu terminal: uvicorn app.main:app --reload Acesse a documentação gerada em http://127.0.0.1:8000/docs.

---

✅ Pré-Requisitos deste Módulo

Antes de expor os endpoints e a segurança da API na internet, certifique-se de que:

• As regras de negócio assíncronas e os métodos transacionais estão implementados nos arquivos de serviço do Módulo 04.
• A dependência pyjwt[crypto] e a biblioteca passlib[bcrypt] estão corretamente instaladas em seu ambiente virtual.

---

🤔 Por que fizemos assim?

• Por que usar a injeção de dependência (Depends) do FastAPI? Ela desacopla o tratamento de requisições de outras responsabilidades. Ao usar Depends(get_db), delegamos a abertura e fechamento da conexão do banco de dados ao framework de forma limpa. Ao injetar Depends(obter_usuario_logado), validamos a assinatura criptográfica do token e extraímos o usuário atual antes mesmo que a lógica do endpoint comece a rodar, simplificando os controladores.
• Por que autenticação stateless por JWT? Tradicionalmente, servidores gravam sessões de login na memória local (stateful). Em infraestruturas modernas na nuvem (como Render), as APIs precisam ser escaláveis e distribuíveis (sem estado). O token JWT carrega as informações do usuário assinadas digitalmente pela chave secreta do servidor, permitindo que a autenticação ocorra sem a necessidade de consultar sessões persistentes no backend.
• Por que configurar o middleware de CORS explicitamente? Navegadores implementam uma política rígida de segurança chamada Same-Origin Policy. Como o frontend do Vue 3 rodará em um provedor (Netlify) e a API FastAPI em outro (Render), o navegador bloqueará as chamadas por padrão se os domínios diferirem. Habilitar o CORSMiddleware injeta os cabeçalhos de controle HTTP necessários informando ao navegador que a origem do frontend é de confiança.

---

🔍 Checkpoint

1. Swagger UI Operante: Execute localmente a aplicação com o uvicorn:

   uvicorn app.main:app --reload
   

   Acesse http://127.0.0.1:8000/docs no navegador e certifique-se de que a página de documentação autogerada carregou todos os endpoints de /produtos, /pedidos e /auth.

2. Bloqueio de Segurança: Tente fazer um POST no endpoint /api/produtos pelo Swagger sem estar autenticado e verifique se a API retorna o status 401 Unauthorized.
3. Botão de Autorização: Verifique se o botão verde de cadeado (“Authorize”) está visível no topo direito do Swagger, permitindo a inserção do token.

---

⚠️ Erros Comuns

Erro	Causa	Solução
Bloqueio por CORS no console de desenvolvedor do navegador	O endereço físico do frontend local ou de produção não foi cadastrado no array origins de app/main.py.	Verifique o console do navegador para identificar qual porta o Vite local está usando (geralmente http://localhost:5173 ou http://localhost:3000) e cadastre o endereço exato com protocolo e porta na whitelist do CORS.
401 Unauthorized mesmo enviando token válido	O token foi enviado bruto ou sem a formatação correta de cabeçalho HTTP de segurança.	O FastAPI espera que o cabeçalho seja enviado no formato Authorization: Bearer <token_jwt>. Certifique-se de manter o prefixo Bearer e um espaço antes da string de token nas requisições.
Sessões abertas e vazamento de conexões	O método get_db não liberou as conexões do pool do banco após o encerramento dos endpoints.	Sempre utilize a sintaxe de gerador assíncrono com blocos de contexto: async with async_session() as session: yield session. O yield garante a suspensão do escopo e a liberação da sessão assim que a requisição retorna.

---

🏁 Conclusão

Incrível! Toda a API REST, o controle de transações assíncronas do banco, a criptografia hash das senhas e a segurança por tokens stateless JWT estão 100% integradas e operando com qualidade profissional.

No Módulo 06, abandonaremos o backend e iniciaremos a construção do nosso cliente SPA em Vue 3! Utilizaremos o Vite para estruturar o esqueleto do projeto TypeScript e criaremos as rotas lógicas da nossa aplicação utilizando o Vue Router.

---

Voltar para o Sumário

⬅️ Anterior Próximo ➡️