Middleware — Interceptando e Aprimorando Requisições e Respostas

Middleware é como o posto de controle de segurança do aeroporto—cada passageiro (requisição) deve passar pela alfândega, segurança e portão de embarque em sequência, e em qualquer uma dessas paradas, o passageiro pode ser autorizado a prosseguir ou parado, enquanto a viagem de volta (resposta) segue a ordem inversa.

1. O Que Você Vai Aprender

• Como o Middleware do Starlette Funciona: Cadeia de Empacotamento ASGI Callable
• Middleware Embutido: Configuração do CORSMiddleware e Políticas de Segurança
• Middleware personalizado: cronometragem de requisição, injeção de ID de requisição, injeção de cabeçalho de resposta
• Ordem de Execução do Middleware: A Relação Entre a Ordem de Registro e a Ordem de Execução Real
• Cenário da Alice: Adicionar middleware de log de requisição e middleware de limitação de taxa da API ao PriceTracker

2. A História Real da Alice

(1) Problema: A API sofre picos de tráfego malicioso e não pode ser rastreada

Depois que Alice lançou sua API PriceTracker, ela descobriu que um determinado endereço IP estava enviando 5.000 requisições por minuto, fazendo com que a carga do banco de dados disparasse. Para piorar, quando o front end do Bob chamou a API de localhost:3000, as requisições foram bloqueadas pela política CORS do navegador, fazendo com que todas falhassem. Alice precisava resolver simultaneamente os problemas de acesso cross-origin e limitação de taxa de requisição, mas o Flask não possui um mecanismo de middleware unificado.

(2) Soluções com Middleware do FastAPI

O FastAPI/Starlette fornece middleware estilo cebola, com cada camada resolvendo um problema específico: o middleware CORS lida com requisições cross-origin, o middleware de limitação de taxa controla a frequência de requisições, e o middleware de log registra requisições—eles não interferem uns nos outros e entram em vigor imediatamente após o registro.

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:3000"])

(3) Resultado

Bob resolveu o problema cross-domain do front end com apenas 5 linhas de código. O middleware de limitação de taxa reduziu as requisições de IPs maliciosos de 5.000 para 1.000 por minuto, e o middleware de log atribuiu um ID único a cada requisição para fácil rastreamento.

3. O Modelo Cebola do Middleware

(1) Princípio do Modelo Cebola

flowchart TD
    Request[Client Request] --> CORS[CORS Middleware]
    CORS --> Logging[Logging Middleware]
    Logging --> RateLimit[Rate Limit Middleware]
    RateLimit --> App[FastAPI Application]
    App --> RateLimit2[Rate Limit Response]
    RateLimit2 --> Logging2[Logging Response]
    Logging2 --> CORS2[CORS Response]
    CORS2 --> Response[Client Response]
    
    RateLimit -.->|429 Too Many Requests| Reject[Short-circuit Rejection]
    Reject --> Logging2

flowchart TD
    Request[Client Request] --> CORS[CORS Middleware]
    CORS --> Logging[Logging Middleware]
    Logging --> RateLimit[Rate Limit Middleware]
    RateLimit --> App[FastAPI Application]
    App --> RateLimit2[Rate Limit Response]
    RateLimit2 --> Logging2[Logging Response]
    Logging2 --> CORS2[CORS Response]
    CORS2 --> Response[Client Response]
    
    RateLimit -.->|429 Too Many Requests| Reject[Short-circuit Rejection]
    Reject --> Logging2

(1) ▶ Exemplo: Ordem de Registro e Ordem de Execução do Middleware

from fastapi import FastAPI, Request
import time

app = FastAPI()

# Middleware registrado PRIMEIRO = camada mais externa (executa primeiro na requisição)
@app.middleware("http")
async def outer_middleware(request: Request, call_next):
    print("Outer: before request")
    response = await call_next(request)
    print("Outer: after response")
    return response

# Middleware registrado POR ÚLTIMO = camada mais interna (executa por último na requisição)
@app.middleware("http")
async def inner_middleware(request: Request, call_next):
    print("Inner: before request")
    response = await call_next(request)
    print("Inner: after response")
    return response

@app.get("/test")
async def test():
    print("Handler: processing")
    return {"ok": True}

Saída:

Outer: before request
Inner: before request
Handler: processing
Inner: after response
Outer: after response

4. CORS Middleware

(1) Configuração de Compartilhamento de Recursos Cross-Domain

CORS (Cross-Origin Resource Sharing) é uma política de segurança do navegador que restringe páginas web de origens diferentes de acessar APIs. Quando o front end do Bob (localhost:3000) acessa a API da Alice (localhost:8000), isso constitui uma requisição cross-origin.

(1) ▶ Exemplo: Configuração CORS para o Ambiente de Desenvolvimento

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Front end do Bob
    allow_credentials=True,
    allow_methods=["*"],  # Todos os métodos HTTP
    allow_headers=["*"],  # Todos os cabeçalhos
)

Saída:

# Execução Bem-sucedida

(2) ▶ Exemplo: Configuração CORS para um Ambiente de Produção

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

ALLOWED_ORIGINS = [
    "https://pricetracker.example.com",
    "https://admin.pricetracker.example.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Saída:

# Execução Bem-sucedida

5. Middleware Personalizado

(1) Middleware de Cronometragem de Requisição

(1) ▶ Exemplo: Registrando o tempo de processamento de cada requisição

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}s"
    return response

@app.get("/products")
async def list_products():
    return [{"id": 1, "name": "Widget"}]

Saída:

# Função definida com sucesso

(2) ▶ Exemplo: Middleware de Injeção de ID de Requisição

import uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id  # Anexar ao estado da requisição
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

@app.get("/health")
async def health_check(request: Request):
    return {
        "status": "healthy",
        "request_id": request.state.request_id,
    }

Saída:

# Função definida com sucesso

(3) ▶ Exemplo: Middleware de Limitação de Taxa da API (Versão Simplificada)

from fastapi import FastAPI, Request, HTTPException
from collections import defaultdict
import time

app = FastAPI()

# Limitador de taxa simples em memória
rate_limits: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT = 1000  # requisições por minuto
WINDOW = 60  # segundos

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host if request.client else "unknown"
    now = time.time()
    
    # Limpar entradas antigas
    rate_limits[client_ip] = [
        t for t in rate_limits[client_ip] if now - t < WINDOW
    ]
    
    # Verificar limite
    if len(rate_limits[client_ip]) >= RATE_LIMIT:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded: {RATE_LIMIT} requests per {WINDOW}s",
        )
    
    rate_limits[client_ip].append(now)
    response = await call_next(request)
    response.headers["X-RateLimit-Limit"] = str(RATE_LIMIT)
    response.headers["X-RateLimit-Remaining"] = str(
        RATE_LIMIT - len(rate_limits[client_ip])
    )
    return response

Saída:

# Função definida com sucesso

6. Middleware Baseado em Classe e ASGI Callables

(1) Como Escrever Middleware Baseado em Classe

Para lógica de middleware mais complexa, recomendamos usar uma abordagem baseada em classe que manipula diretamente o scope, receive e send do ASGI.

(1) ▶ Exemplo: Middleware de Log de Requisição Baseado em Classe

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import logging
import time

logger = logging.getLogger("pricetracker")

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        start = time.perf_counter()
        logger.info(f"Request: {request.method} {request.url.path}")
        
        response = await call_next(request)
        
        duration = time.perf_counter() - start
        logger.info(
            f"Response: {request.method} {request.url.path} "
            f"status={response.status_code} duration={duration:.4f}s"
        )
        return response

# Registrar middleware baseado em classe
app = FastAPI()
app.add_middleware(LoggingMiddleware)

Saída:

# Função definida com sucesso

(2) Comparação de Tipos de Middleware

❓ Perguntas Frequentes

📖 Resumo

• O middleware segue o modelo cebola: requisições viajam de fora para dentro, e respostas viajam de dentro para fora; qualquer camada pode retornar uma resposta antecipadamente
• CORSMiddleware para resolver problemas cross-domain, você deve especificar um nome de domínio específico no ambiente de produção
• Middleware personalizado pode implementar preocupações transversais como cronometragem de requisição, injeção de ID de requisição e limitação de taxa
• A ordem em que o middleware é registrado determina a ordem de execução: o último a ser registrado é o primeiro a processar requisições (camada mais interna)
• Middleware baseado em classe (BaseHTTPMiddleware) é adequado para lógica complexa; classes ASGI puras podem lidar com WebSockets

📝 Exercícios

1. Problema Básico (Dificuldade ⭐): Adicione um middleware CORS ao PriceTracker para permitir acesso cross-origin de http://localhost:3000, e verifique em um navegador que o front end do Bob pode chamar a API. Dica: app.add_middleware(CORSMiddleware, ...)
2. Exercício Avançado (Dificuldade: ⭐⭐): Implemente um middleware de cronometragem de requisição que adicione X-Process-Time aos cabeçalhos de resposta, e use o Swagger UI para visualizar os cabeçalhos de resposta. Dica: @app.middleware("http") + time.perf_counter()
3. Desafio (Dificuldade: ⭐⭐⭐): Implemente middleware de limitação de taxa baseado em IP (1.000 requisições por minuto). Quando o limite for excedido, retorne um status 429 e o cabeçalho de resposta Retry-After, enquanto exibe a cota restante no cabeçalho de resposta. Dica: defaultdict(list) + limpeza de janela de tempo

---|

Equipe Técnica Web-Tutorial

Uma plataforma de tutoriais mantida por diversos desenvolvedores. Cada tutorial é escrito e revisado por profissionais da área correspondente. Trabalhamos para manter nosso conteúdo preciso e confiável — se encontrar algum problema, avise-nos.

Sobre Nós