💻 Exemplos: Capítulo 12: Middlewares Customizados, Tratamento Global de Exceções e CORS
Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis
💻 Exemplo 1 — Middlewares Encadeados: Tempo de Processamento + Correlation ID
import time
import uuid
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def correlation_id_middleware(request: Request, call_next):
correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
request.state.correlation_id = correlation_id
response = await call_next(request)
response.headers["X-Correlation-ID"] = correlation_id
return response
@app.middleware("http")
async def process_time_middleware(request: Request, call_next):
inicio = time.perf_counter()
response = await call_next(request)
duracao = time.perf_counter() - inicio
response.headers["X-Process-Time"] = f"{duracao:.4f}"
return response
@app.get("/status")
async def status_endpoint(request: Request):
return {"status": "ok", "correlation_id": request.state.correlation_id}
💻 Exemplo 2 — Exceções de Domínio Traduzidas por exception_handler Global
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class DominioException(Exception):
"""Classe-base para todas as exceções de regra de negócio da aplicação."""
class RecursoNaoEncontradoException(DominioException):
def __init__(self, entidade: str, identificador: int) -> None:
self.mensagem = f"{entidade} com id={identificador} não encontrado(a)"
class RegraDeNegocioVioladaException(DominioException):
def __init__(self, mensagem: str) -> None:
self.mensagem = mensagem
app = FastAPI()
@app.exception_handler(RecursoNaoEncontradoException)
async def handler_recurso_nao_encontrado(request: Request, exc: RecursoNaoEncontradoException):
return JSONResponse(status_code=404, content={"erro": "RECURSO_NAO_ENCONTRADO", "detalhe": exc.mensagem})
@app.exception_handler(RegraDeNegocioVioladaException)
async def handler_regra_negocio(request: Request, exc: RegraDeNegocioVioladaException):
return JSONResponse(status_code=422, content={"erro": "REGRA_NEGOCIO_VIOLADA", "detalhe": exc.mensagem})
PEDIDOS = {1: {"id": 1, "status": "ABERTO"}}
@app.get("/pedidos/{pedido_id}")
async def buscar_pedido(pedido_id: int):
if pedido_id not in PEDIDOS:
raise RecursoNaoEncontradoException("Pedido", pedido_id)
return PEDIDOS[pedido_id]
@app.post("/pedidos/{pedido_id}/cancelar")
async def cancelar_pedido(pedido_id: int):
pedido = PEDIDOS.get(pedido_id)
if pedido is None:
raise RecursoNaoEncontradoException("Pedido", pedido_id)
if pedido["status"] == "ENVIADO":
raise RegraDeNegocioVioladaException("Pedidos já enviados não podem ser cancelados")
pedido["status"] = "CANCELADO"
return pedido
💻 Exemplo 3 — CORS Restrito por Origem em Ambiente de Produção
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
ORIGENS_PERMITIDAS = [
"https://app.minhaempresa.com",
"https://admin.minhaempresa.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=ORIGENS_PERMITIDAS,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
expose_headers=["X-Process-Time", "X-Correlation-ID"],
)
@app.get("/saude")
async def verificar_saude():
return {"status": "operacional"}
# Requisição de https://app.minhaempresa.com -> navegador libera (origem na lista)
# Requisição de https://site-desconhecido.com -> navegador bloqueia no preflight,
# pois a origem não está em ORIGENS_PERMITIDAS e allow_origins=["*"] jamais
# deveria ser combinado com allow_credentials=True (combinação inválida na spec CORS).
print("Middleware CORS registrado para:", ORIGENS_PERMITIDAS)