Camada HTTP

Middlewares, dependências, routers e composição de middleware para a superfície da API.

Você vai montar aqui a superfície HTTP inteira de um serviço a partir dos primitivos que o tempest_fastapi_sdk entrega — sem escrever middleware, exception handler ou glue de bootstrap na mão. Cada seção é independente: pegue só a que você precisa agora. Esta receita cobre ~12 primitivos:

• create_app() + register_exception_handlers — bootstrap canônico e o envelope de erro padronizado (com i18n opcional via MessageCatalog).
• RequestIDMiddleware — correlação X-Request-ID em cada linha de log.
• apply_cors — CORS a partir de CORSSettings.
• make_health_router / make_token_dependency — liveness/readiness + o guarda de segredo compartilhado X-Token.
• Dependências JWT / bearer / role / permission — controle de rota por token e por papel.
• RateLimitMiddleware — janela deslizante, chave por IP/usuário/tenant, store em memória ou Redis.
• make_rate_limit_dependency — o mesmo limite em uma rota só, com chave que pode ler o corpo (key_by_body_field).
• BodySizeLimitMiddleware — teto de bytes no corpo do request, com 413 antes de qualquer parse.
• WebhookSignatureVerifier / RSAWebhookSignatureVerifier — validação de webhooks assinados (HMAC ou RSA).
• build_pagination_link_header — header Link RFC 8288 no estilo GitHub.
• make_tool_spec_router — manifesto legível por máquina no prefixo raiz.
• run_server — ponto de entrada programático do uvicorn.
• BaseAppSettings + mixins *Settings — configuração componível por env var.

As três últimas seções são flows completos

Autenticação, upload e e-mail transacional aparecem aqui em forma resumida; cada uma tem uma receita dedicada e mais profunda — veja o Recap no fim da página.

Bootstrap da aplicação

A seção 2 do tutorial mostra o create_app() mínimo. Esta receita é a versão estendida, conectando tudo que tempest_fastapi_sdk.api entrega — exception handlers, CORS, middleware de request-ID, o health router com checks extras, uma dependência de token de segredo compartilhado e um manager extra de Redis — tudo a partir da mesma localização canônica src/api/app.py. O padrão de bootstrap continua idêntico; só o conteúdo de create_app() cresce.

# src/api/app.py
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI

from tempest_fastapi_sdk import (
    AsyncDatabaseManager,
    RequestIDMiddleware,
    apply_cors,
    configure_logging,
    make_health_router,
    make_token_dependency,
    register_exception_handlers,
)
from tempest_fastapi_sdk.cache import AsyncRedisManager

from src.core.settings import settings


configure_logging(level=settings.LOG_LEVEL, json_output=settings.LOG_JSON)

db = AsyncDatabaseManager(
    settings.DATABASE_URL,
    echo=settings.DATABASE_ECHO,
    pool_size=settings.DATABASE_POOL_SIZE,
    max_overflow=settings.DATABASE_MAX_OVERFLOW,
    pool_recycle=settings.DATABASE_POOL_RECYCLE,
)
redis = AsyncRedisManager(settings.REDIS_URL)
require_token = make_token_dependency(settings.TOKEN_SECRET)


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    await db.connect()
    await redis.connect()
    try:
        yield
    finally:
        await redis.disconnect()
        await db.disconnect()


def create_app() -> FastAPI:
    """Build and configure the FastAPI app."""
    app = FastAPI(
        title="my-service",
        version=settings.VERSION,
        lifespan=lifespan,
    )

    app.add_middleware(RequestIDMiddleware)
    apply_cors(app, settings)
    register_exception_handlers(app)

    # Meta endpoints at the root prefix.
    app.include_router(
        make_health_router(
            db=db,
            checks={"redis": redis.health_check},
            version=settings.VERSION,
        ),
    )

    # Business endpoints under /api/<domain>, guarded by the shared secret.
    from src.api.routers import users

    app.include_router(
        users.router,
        prefix="/api",
        dependencies=[Depends(require_token)],
    )
    return app


app = create_app()

Pontos-chave:

• src/server.py e main.py (one-liner) ficam exatamente como na seção 2 do tutorial — só create_app() muda quando você adiciona primitivos. Nunca inicie o uvicorn via subprocess.run(["uvicorn", ...]); sempre importe app de src.api.app ou chame uvicorn.run("src.api.app:app", ...) programaticamente de src/server.py.
• RequestIDMiddleware lê/escreve X-Request-ID e semeia request_id_ctx para que toda linha de log emitida durante a requisição carregue o ID de correlação.
• apply_cors(app, settings) lê os defaults de CORSSettings; passe overrides nomeados para mudanças pontuais.

• register_exception_handlers(app) conecta três handlers, cada um com seu nível de log (mais um quarto, para o 422, quando envelope_validation_errors=True — veja O 422 dentro do envelope):

  • AppException → envelope {detail, code, details} + log INFO (4xx) ou ERROR + traceback + 500.log (5xx).
  • HTTPException → mantém o body padrão do Starlette ({"detail"}) em 4xx com log INFO; em 5xx aplica o envelope SDK + traceback + 500.log.
  • Exception (catch-all) → envelope SDK + traceback + 500.log (corrige o default do Starlette, que devolve só "Internal Server Error" sem log).

  Todos os handlers respeitam RequestIDMiddleware: a linha de log carrega o request_id, e o envelope expõe ele em details para correlacionar com o cliente. Passe log_traceback=False se um APM (Sentry, OpenTelemetry) já estiver capturando a trace. - make_health_router(db=db, checks={"redis": redis.health_check}, version=...) monta GET /health/liveness e GET /health/readiness (retorna 503 quando algum check falha) no prefixo raiz — veja Liveness e readiness. - make_token_dependency(secret) retorna uma dependência async que valida X-Token via hmac.compare_digest; passe uma string vazia para desabilitar no dev. A dependência vive ao lado do resto da cola de auth em src/api/dependencies/auth.py quando crescer além do one-liner acima.

Liveness e readiness

São duas perguntas diferentes, feitas por quem decide coisas diferentes:

• Liveness — "o processo está vivo?". Se falha, o orquestrador (Kubernetes, ECS) reinicia o container. Por isso GET /health/liveness não toca dependência nenhuma: um Redis fora do ar não se resolve reiniciando a API, e um liveness que dependesse dele derrubaria todas as réplicas ao mesmo tempo, em loop.
• Readiness — "posso receber tráfego agora?". Se falha, o balanceador tira a réplica da rotação e devolve quando voltar a passar. É aqui que entram banco, cache e broker: GET /health/readiness roda cada check e responde 503 quando algum falha.

Um único GET /health mistura as duas coisas — ou reinicia por causa de uma dependência, ou manda tráfego para quem não consegue atendê-lo. Ao pedir o health de um serviço novo, peça os dois endpoints.

import asyncio

from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient

from tempest_fastapi_sdk import make_health_router


async def redis_check() -> bool:
    """Responde na hora: dependência saudável."""
    return True


async def search_check() -> bool:
    """Pendura, como um socket meio aberto que nunca responde."""
    await asyncio.sleep(60)
    return True


app = FastAPI()
app.include_router(
    make_health_router(
        checks={"redis": redis_check, "search": search_check},
        timeout=0.5,
    ),
)


async def main() -> None:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        liveness = await client.get("/health/liveness")
        readiness = await client.get("/health/readiness")
    print(liveness.status_code, liveness.json())
    print(readiness.status_code, readiness.json())


asyncio.run(main())

Saída (o processo inteiro levou 1,7 s, apesar do sleep(60)):

Health check 'search' timed out after 0.5s
200 {'status': 'ok'}
503 {'status': 'not_ready', 'checks': {'redis': True, 'search': False}}

O que acontece na readiness:

• Os checks rodam em paralelo, cada um limitado por timeout (default 3.0 segundos; None desliga). O que estoura conta como falha só para ele — o redis acima continua True —, e a resposta sai logo depois do timeout, não quando o probe do orquestrador desiste. Mantenha o timeout abaixo do timeout do probe.
• O log não leva a mensagem da exceção. Um check que levanta gera Health check 'database' raised ConnectionRefusedError — nome e tipo, nunca o texto, porque mensagem de driver de banco ou de broker costuma carregar a DSN com usuário e senha.

Cancelamento só acontece num await

O timeout cancela o check no próximo ponto de await. Um check que bloqueia o event loop com I/O síncrono (um driver não-async, um time.sleep) não pode ser interrompido — e trava o processo inteiro, não só a readiness. Rode esse tipo de chamada com asyncio.to_thread.

Fila

MessageBroker.health_check faz o ping real do FastStream, com timeout=2.0 por default — abaixo dos 3 s do router. Com o RabbitMQ parado depois do boot, ele devolve False: make_health_router(checks={"queue": mq.health_check}).

Todo 4xx dentro do envelope

O 422 era o furo mais visível, não o único. As outras quatro categorias de erro de cliente passam pelo handler do SDK — seis casos, todos medidos na tabela abaixo — e respondiam o corpo nu do Starlette, sem code e em inglês, mesmo num serviço com MessageCatalog configurado:

Caminho	Antes	Com envelope_client_errors=True
raise HTTPException(404, "pedido 42 não existe")	{"detail": "pedido 42 não existe"}	mesma mensagem + code: NOT_FOUND
raise HTTPException(403)	{"detail": "Forbidden"}	Acesso negado + code: FORBIDDEN
raise HTTPException(409)	{"detail": "Conflict"}	Conflito de recurso + code: CONFLICT
Depends(HTTPBearer()) sem token	{"detail": "Not authenticated"}	mesma mensagem + code: UNAUTHORIZED
rota inexistente	{"detail": "Not Found"}	Recurso não encontrado + code: NOT_FOUND
método errado	{"detail": "Method Not Allowed"}	mesma mensagem + code: HTTP_405

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import default_message_catalog, register_exception_handlers


def create_app() -> FastAPI:
    """Monta a aplicação.

    Returns:
        FastAPI: A aplicação configurada.
    """
    app: FastAPI = FastAPI()
    register_exception_handlers(
        app,
        catalog=default_message_catalog(),
        default_locale="pt-BR",
        envelope_client_errors=True,       # ← todo 4xx, o 422 incluído
    )
    return app

Mensagem que você escreveu não é traduzida

Essa é a linha que separa "localizar" de "perder informação". O Starlette preenche detail com a phrase do status quando o caller não passa uma, então os dois casos são distinguíveis:

• HTTPException(404, "pedido 42 não existe") → mensagem sua, preservada
• HTTPException(404) → detail == "Not Found", que é a phrase → traduzida

Trocar a primeira pelo genérico do catálogo destruiria a mensagem mais informativa das duas. O "Not authenticated" do HTTPBearer cai no primeiro grupo também — medido — e é preservado.

Por que dois flags e não um renomeado

envelope_validation_errors shippou na v0.284.0 cobrindo só o 422. envelope_client_errors é o superconjunto: liga todo 4xx, o 422 incluído. Quem já tinha ligado o primeiro não muda de comportamento ao atualizar — que é exatamente o que um rename silencioso faria.

Precisa de tudo? Só envelope_client_errors=True. Precisa só do 422? Só o outro.

HTTP_<status> para status que o SDK não modela

O mapa status→code é declarado, não derivado: medido, treze classes de AppException carregam um code de 401 e quatro de 404, então não há status de onde ler um code único.

Status fora do mapa responde HTTP_<status> e mantém a phrase do framework. 400 e 405 estão nesse grupo hoje — nenhuma subclasse de AppException carrega code para eles. O cliente ainda parseia um formato só; o que falta é a tradução, não o code.

O 422 dentro do envelope

register_exception_handlers conecta três handlers, e o 422 não passa por nenhum: o FastAPI intercepta RequestValidationError no handler default dele. Então o erro que o cliente vê mais era o único fora do modelo de erro do SDK — sem code, e sem passar pelo MessageCatalog que o serviço configurou.

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import default_message_catalog, register_exception_handlers


def create_app() -> FastAPI:
    """Monta a aplicação.

    Returns:
        FastAPI: A aplicação configurada.
    """
    app: FastAPI = FastAPI()
    register_exception_handlers(
        app,
        catalog=default_message_catalog(),
        default_locale="pt-BR",
        envelope_validation_errors=True,        # ← liga o 422 no envelope
    )
    return app

Com o flag ligado, POST /login com password curta responde:

{
  "detail": "Erro de validação",
  "code": "VALIDATION_ERROR",
  "details": {
    "errors": [
      {
        "loc": ["body", "password"],
        "type": "string_too_short",
        "msg": "O texto deve ter no mínimo 12 caractere(s)"
      }
    ]
  }
}

O 422 default devolve o valor submetido — a senha inclusive

O corpo que o FastAPI gera inclui input para cada campo que falhou. Medido, com password reprovando em min_length:

{"detail": [{"type": "string_too_short", "loc": ["body", "password"],
             "msg": "String should have at least 12 characters",
             "input": "hunter2"}]}

E SecretStr não protege: a validação roda antes de o wrapper de segredo existir, então um campo SecretStr que reprova também devolve o valor cru. Medido nos dois casos.

Esse é o motivo mais forte para ligar o flag. details carrega loc e type, e nunca input — não é um aviso para você implementar, é uma linha que o handler não escreve.

A tabela é por tipo de erro, não por campo

As mensagens saem do catálogo com a chave VALIDATION.<tipo do pydantic> — string_too_short, missing, int_parsing e as outras 101. Isso é vocabulário do pydantic, igual em todo serviço, então campo novo num schema nasce traduzido: não há tabela por campo para manter.

São 104 tipos, o conjunto exato do pydantic_core.ErrorType, e o tests/test_pydantic_error_types_guard.py falha quando o pydantic instalado e a tabela discordam em qualquer direção.

O catálogo carrega só português, de propósito: o msg do próprio pydantic já é a mensagem em inglês, então o handler cai nela e o SDK não mantém uma cópia que pode divergir do upstream. Tipo que o catálogo não conhece também cai no msg — degrada para inglês, nunca para branco.

Os validadores BR do SDK localizam a frase inteira

O pydantic arquiva tudo que um validador de campo levanta sob o tipo único value_error, com a exceção em ctx["error"]. Como o catálogo localiza por tipo, CPFOrCNPJField, PhoneBRField, UFField, CEP e chave PIX chegavam pela metade:

Valor inválido: invalid CPF/CNPJ

Desde a v0.290.0 esses validadores levantam ValidationValueError, que carrega um código (INVALID_CPF_CNPJ, INVALID_CEP, INVALID_UF, …) resolvido como VALIDATION.<código> nas duas línguas — CPF ou CNPJ inválido em pt-BR, Invalid CPF or CNPJ em en-US. Aqui o inglês é carregado, ao contrário da tabela do pydantic: o msg do upstream para esse caso é Value error, invalid CPF/CNPJ, prefixado e em minúscula.

O seu validador não muda: um ValueError comum continua caindo em VALIDATION.value_error, e ValidationValueError é subclasse de ValueError, então todo except ValueError segue pegando.

Ligar o flag muda o corpo, então é opt-in

Frontend que faz body.detail.map(...) e cliente gerado do OpenAPI modelam o 422 como array. Por isso o default é False, e ligar é uma decisão do serviço.

O schema acompanha: com o flag ligado, o componente HTTPValidationError é reescrito para o envelope (detail como string, code, details) e o item perde input. Todas as rotas referenciam esse componente, então a doc e o runtime concordam nas duas posições do flag — sem isso, o schema mentiria exatamente na resposta que o cliente gerado mais usa.

Avisando alguém, e escolhendo para onde o registro vai

log_level decide a severidade do 5xx. Duas coisas que ele não decide: quem fica sabendo, e em qual configuração de logging o registro cai.

# src/api/app.py

import logging

from fastapi import FastAPI, Request

from tempest_fastapi_sdk import EmailUtils, LogUtils, register_exception_handlers

from src.core.settings import settings

log: LogUtils = LogUtils(__name__, scope="logger")


async def alert_ops(request: Request, exc: Exception) -> None:
    """Avisa a operação de um 5xx, fora do caminho da resposta.

    Args:
        request (Request): A requisição que falhou.
        exc (Exception): A exceção relatada.
    """
    mailer: EmailUtils = EmailUtils(**settings.email_kwargs())
    await mailer.send(
        to=[settings.OPS_EMAIL],
        subject=f"5xx em {request.url.path}",
        body=f"{type(exc).__name__}: {exc}",
    )


def create_app() -> FastAPI:
    """Monta a aplicação.

    Returns:
        FastAPI: A aplicação configurada.
    """
    app: FastAPI = FastAPI()
    register_exception_handlers(
        app,
        logger=log.logger,                # ← onde o registro cai
        on_server_error=alert_ops,        # ← quem fica sabendo
        log_level=logging.ERROR,
    )
    return app

on_server_error roda como background task

Chamado com (request, exc) depois de a resposta estar montada, então nunca atrasa nem altera o que o cliente recebe. Dispara nos três caminhos que produzem 5xx — o catch-all, HTTPException(500) e AppException de 5xx —, e não dispara em 4xx: alerta que dispara em 404 é alerta que ninguém lê.

Exceção dentro do callback é logada e engolida

De propósito. Medido sem essa proteção: a resposta 500 sai correta, mas a exceção do notificador propaga pela pilha ASGI e passa a ser o que o servidor relata — o 500 do cliente vira um 500 pior, sem rastro da causa real. Com a proteção, o que sobe é a exceção original.

logger= importa quando o logging é de escopo isolado

Os handlers logam no logger deles (tempest_fastapi_sdk.api.handlers). Com LogUtils(..., scope="root") — o default — isso já cai no 500.log e no error.log do serviço; medido, 1 linha em cada.

Com scope="logger", que não configura a raiz, o logger do SDK fica fora da árvore configurada e o registro não chega a nenhum dos arquivos — medido, 0 e 0. Passar logger=log.logger fecha isso: 1 e 1.

O tipo é logging.Logger, e não o protocolo RetryLogger, porque os handlers passam extra= e exc_info= e o protocolo não tem nenhum dos dois. Medido: LogUtils.error(extra=...) guarda um campo chamado extra, e logging.Logger.error(request_id=...) levanta TypeError — nenhum tipo único cobre os dois. LogUtils expõe .logger, que é o objeto que serve.

Mensagens de erro localizadas (i18n)

Por padrão o detail do envelope é a mensagem literal da exceção (em inglês nos built-ins). Para devolver a mensagem no idioma do cliente sem traduzir em cada raise, passe um MessageCatalog para register_exception_handlers:

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import default_message_catalog, register_exception_handlers


def create_app() -> FastAPI:
    app = FastAPI(...)
    register_exception_handlers(
        app,
        catalog=default_message_catalog(),                   # ← PT-BR + EN-US embutidos
        default_locale="pt-BR",
    )
    ...

O handler negocia o locale a partir do header Accept-Language (ordenado por q), cai em default_locale quando nada casa, e resolve a chave da exceção — message_key se definido, senão o code — contra o catálogo. Sem catálogo, ou quando a chave não existe, mantém o detail literal (zero quebra de compatibilidade).

# Mesmo NotFoundException, idioma decidido pelo Accept-Language do cliente:
#   Accept-Language: pt-BR  →  {"detail": "Recurso não encontrado", "code": "NOT_FOUND"}
#   Accept-Language: en-US  →  {"detail": "Resource not found",     "code": "NOT_FOUND"}

Para códigos de domínio (e mensagens com parâmetros), estenda o catálogo com merge e passe message_params no raise:

# src/core/i18n.py
from tempest_fastapi_sdk import MessageCatalog, default_message_catalog

CATALOG: MessageCatalog = default_message_catalog().merge(
    {
        "pt-BR": {"USER_NOT_FOUND": "Usuário {email} não encontrado"},
        "en-US": {"USER_NOT_FOUND": "User {email} not found"},
    }
)

# src/services/user.py
from tempest_fastapi_sdk import NotFoundException


def require_user(email: str) -> None:
    """Raise a localized 404 carrying the offending e-mail.

    Args:
        email (str): The e-mail that was not found.

    Raises:
        NotFoundException: Always — keyed to ``USER_NOT_FOUND`` so the
            handler localizes it from the request locale.
    """
    raise NotFoundException(
        "User not found",                                    # fallback literal
        code="USER_NOT_FOUND",
        message_params={"email": email},
    )

A chave segue o code por padrão

Você raramente passa message_key — ele cai no code da exceção. Defina message_key só quando quiser desacoplar a string traduzida do código de erro. Um template que referencia um parâmetro ausente volta sem interpolar, em vez de estourar.

Dependências JWT bearer / usuário atual / role

Quatro factories de dependência vivem em tempest_fastapi_sdk.api.dependencies.auth — escolha o nível de abstração que você precisa.

Factory	O que você ganha
make_token_dependency(secret)	Valida o header de segredo compartilhado X-Token (tempo constante).
make_bearer_token_dependency(tokens, soft=False)	Decodifica Authorization: Bearer <jwt> e retorna o dict de claims.
make_jwt_user_dependency(tokens, user_loader, soft=False, subject_claim="sub")	Decodifica o bearer JWT, aguarda user_loader(subject), retorna o usuário carregado.
make_role_dependency(tokens, ["admin"], require_all=False, roles_claim="roles") / make_permission_dependency(tokens, ["users:write"], require_all=True, permissions_claim="permissions")	Decodifica o bearer JWT e controla a rota por roles / permissões.

Fora de rota: require_x_token

A checagem do segredo compartilhado também existe em forma imperativa — require_x_token(secret, token) levanta UnauthorizedException quando não casa (e é no-op com secret="", o modo dev). Serve onde não há Depends pra pendurar: um consumidor de fila que recebe o token no payload, um handler de WebSocket, um script. Em rota, prefira make_token_dependency(secret) — o Swagger mostra o header.

from tempest_fastapi_sdk import require_x_token

from src.core.settings import settings


def handle_job(payload: dict[str, str]) -> None:
    """Reject the job unless it carries the shared secret."""
    require_x_token(settings.TOKEN_SECRET, payload.get("token", ""))

Usa o flow bundled? Pule o load_user

Se você monta auth com UserAuthService + make_auth_router, não precisa escrever load_user nem instanciar um JWTUtils aqui — chame auth_service.current_user_dependency() (e .current_user_dependency(soft=True)), que reusa o JWTUtils interno do service. Veja a receita de auth ». O exemplo abaixo é a montagem manual, pra quando você não usa o service.

# src/api/dependencies/auth.py
from uuid import UUID

from tempest_fastapi_sdk import (
    JWTUtils,
    make_bearer_token_dependency,
    make_jwt_user_dependency,
    make_permission_dependency,
    make_role_dependency,
)

from src.api.app import db
from src.core.settings import settings
from src.db.models import UserModel
from src.db.repositories import UserRepository


tokens = JWTUtils(
    secret=settings.JWT_SECRET,
    algorithm=settings.JWT_ALGORITHM,
)


async def load_user(subject: str) -> UserModel:
    """Resolve the JWT subject (a UUID string) to a persisted user."""
    async with db.get_session_context() as session:
        repo = UserRepository(session)
        return await repo.get_by_id(UUID(subject))


require_bearer = make_bearer_token_dependency(tokens)
get_current_user = make_jwt_user_dependency(tokens, load_user)
get_current_user_or_none = make_jwt_user_dependency(tokens, load_user, soft=True)

require_admin = make_role_dependency(tokens, ["admin"])
require_users_write = make_permission_dependency(tokens, ["users:write"])

# src/api/routers/users.py

from uuid import UUID

from fastapi import APIRouter, Depends

from src.api.dependencies.auth import (
    get_current_user,
    require_admin,
    require_users_write,
)
from src.db.models import UserModel
from src.schemas import UserResponseSchema


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


@router.get("/me")
async def me(current: UserModel = Depends(get_current_user)) -> UserResponseSchema:
    return UserResponseSchema.model_validate(current)


@router.delete("/{user_id}", dependencies=[Depends(require_admin)])
async def delete_user(user_id: UUID) -> None:
    ...


@router.patch(
    "/{user_id}/permissions",
    dependencies=[Depends(require_users_write)],
)
async def update_perms(user_id: UUID) -> None:
    ...

soft=True retorna None em vez de levantar em tokens ausentes/inválidos — útil para endpoints que funcionam tanto autenticados quanto anônimos. subject_claim é "sub" por padrão, mas pode ser qualquer claim custom ("user_id", "uid", ...). As dependências de role aceitam uma string ou uma lista de strings no claim do JWT; require_all=True exige cada role/permissão listada, False (default para roles, sobrescrito para permissões) exige qualquer uma.

JWTUtils criado no create_app

Com get_settings() e create_app(settings), o JWTUtils só existe dentro do create_app — e a dependency de módulo (CurrentUser = Annotated[...]) é declarada antes. Passe uma função que recebe o Request no lugar do helper: ela é chamada a cada request. Vale para make_jwt_user_dependency, make_bearer_token_dependency, make_role_dependency e make_permission_dependency.

from typing import Annotated

from fastapi import Depends, FastAPI, Request

from tempest_fastapi_sdk import JWTUtils, make_jwt_user_dependency


async def load_user(subject: str) -> dict[str, str]:
    """Load the user the token names."""
    return {"id": subject}


def jwt_of(request: Request) -> JWTUtils:
    """Read the helper create_app put on app.state."""
    jwt: JWTUtils = request.app.state.jwt
    return jwt


CurrentUser = Annotated[
    dict[str, str],
    Depends(make_jwt_user_dependency(jwt_of, user_loader=load_user, strict=True)),
]


def create_app(secret: str) -> FastAPI:
    """Build the app; the JWT helper only exists from here on."""
    app = FastAPI()
    app.state.jwt = JWTUtils(secret)

    @app.get("/me")
    async def me(user: CurrentUser) -> dict[str, str]:
        """Return the authenticated user."""
        return user

    return app

Medido em tests/api/test_lazy_dependencies.py: dois apps no mesmo processo, com segredos diferentes e a mesma CurrentUser, cada um recusa (401) o token do outro.

Middleware de rate limit

RateLimitMiddleware é um limitador de janela deslizante — cada chave única (IP do cliente por padrão) é permitida no máximo max_requests requisições dentro de cada janela window_seconds. Requisições que excedem ganham um 429 Too Many Requests com header Retry-After e o envelope de erro canônico do SDK no corpo (veja abaixo). Dois eixos são plugáveis: o store (memória ou Redis) e a chave (IP, usuário, tenant, API key) — veja abaixo.

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import RateLimitMiddleware


def create_app() -> FastAPI:
    app = FastAPI(...)
    app.add_middleware(
        RateLimitMiddleware,
        max_requests=120,
        window_seconds=60.0,
        exempt_paths=("/health/liveness", "/health/readiness"),
    )
    ...

Limite por usuário / tenant / API key

Por padrão a chave é o IP do cliente. Para limitar por principal (usuário autenticado, tenant, API key), passe um key_func. O SDK traz factories prontas:

Factory	Chave gerada	Uso
key_by_ip(trusted_header=...)	ip:<addr>	Por IP (default).
key_by_jwt_subject(jwt)	user:<sub>	Por usuário autenticado (claim sub).
key_by_jwt_claim(jwt, "tenant_id", scope="tenant")	tenant:<id>	Por claim arbitrária do token.
key_by_header("x-api-key", scope="apikey")	apikey:<valor>	Por valor de header.

O middleware roda antes das dependencies

O RateLimitMiddleware executa antes das Depends do FastAPI resolverem — então o usuário autenticado pela sua dependency de auth ainda não existe quando a chave é calculada. Por isso as factories key_by_jwt_* decodificam o bearer do request cru (via JWTUtils.decode_or_none, sem levantar exceção). Tráfego anônimo cai de volta no IP, então continua limitado.

# src/api/app.py
from fastapi import FastAPI

from tempest_fastapi_sdk import RateLimitMiddleware, key_by_jwt_subject

from src.api.dependencies.resources import get_jwt_utils


def create_app() -> FastAPI:
    app = FastAPI(...)
    app.add_middleware(
        RateLimitMiddleware,
        max_requests=600,
        window_seconds=60.0,
        key_func=key_by_jwt_subject(get_jwt_utils()),        # ← limite por usuário
        exempt_paths=("/health/liveness", "/health/readiness"),
    )
    return app

Estado distribuído com Redis

O store padrão (MemoryRateLimitStore) conta em processo — correto para um único worker. Para deploys multi-réplica, passe store=RedisRateLimitStore(redis): cada chave vira um sorted set e um único script Lua poda os expirados, conta e adiciona o novo hit atomicamente (sem corrida entre contar e adicionar). Em erro do Redis, fail_open=True (default) libera a requisição em vez de derrubar todo mundo.

# src/api/app.py

from fastapi import FastAPI
from redis.asyncio import Redis

from tempest_fastapi_sdk import (
    RateLimitMiddleware,
    RedisRateLimitStore,
    key_by_jwt_subject,
)

from src.api.dependencies.resources import get_jwt_utils
from src.core.settings import settings


def create_app() -> FastAPI:
    redis: Redis = Redis.from_url(settings.REDIS_URL)
    app = FastAPI(...)
    app.add_middleware(
        RateLimitMiddleware,
        max_requests=600,
        window_seconds=60.0,
        key_func=key_by_jwt_subject(get_jwt_utils()),
        store=RedisRateLimitStore(redis),                    # ← compartilhado entre réplicas
        exempt_paths=("/health/liveness", "/health/readiness"),
    )
    return app

A semântica de janela deslizante é idêntica nos dois stores; só muda onde os contadores vivem. Ainda dá para empurrar o rate limiting para a borda (nginx / Cloudflare / AWS WAF) quando preferir.

Por que Redis.from_url aqui, e não AsyncRedisManager?

Este client alimenta um middleware, montado no create_app (síncrono), antes de qualquer lifespan async rodar. Redis.from_url() é lazy — constrói sem abrir conexão, então serve nesse ponto. O AsyncRedisManager exige await connect() e cabe onde há contexto async: client via Depends(cache.client_dependency), ou o SSEBroker montado no lifespan. Os dois precisam do extra [cache] (o pacote redis).

Rajada tolerada: token bucket

A janela deslizante responde uma pergunta só: essa chave passou de N requisições nos últimos W segundos? Um cliente que dispara 20 requisições de uma vez e depois fica um minuto quieto é bem-comportado na média — e mesmo assim a janela estrita o rejeita.

O token bucket separa as duas coisas: a taxa sustentada (tokens repostos por segundo) e a rajada que ele absorve (capacidade do balde). Uma RateLimitRule com burst vira token bucket; sem burst, continua janela deslizante.

from tempest_fastapi_sdk import RateLimitRule

# 10 req/s sustentadas (600 por minuto), absorvendo rajada de 100.
rule = RateLimitRule(max_requests=600, window_seconds=60.0, burst=100)

Regras vão para o middleware através de uma policy. Para um serviço de tier único, StaticRateLimitPolicy aplica a mesma lista a todo mundo:

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import (
    RateLimitMiddleware,
    RateLimitRule,
    StaticRateLimitPolicy,
)


def create_app() -> FastAPI:
    app = FastAPI()
    app.add_middleware(
        RateLimitMiddleware,
        policy=StaticRateLimitPolicy(
            [RateLimitRule(max_requests=600, window_seconds=60.0, burst=100)],
        ),
        exempt_paths=("/health/liveness", "/health/readiness"),
    )
    return app

policy= e store= são alternativas

max_requests / window_seconds / store descrevem uma janela deslizante. Com policy= você passa uma lista de limites, e os contadores vivem em quota_store= (MemoryQuotaStore por padrão, RedisQuotaStore para multi-réplica). Passar os dois levanta ValueError quando o Starlette monta a pilha de middleware — no startup do lifespan, não na linha do add_middleware, que retorna sem erro. Honrar um significaria ignorar o outro em silêncio.

Cotas por plano

Uma lista de regras por tier, resolvida por requisição. As regras de um plano são checadas juntas: um limite por minuto embaixo de um teto diário.

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import (
    PlanRateLimitPolicy,
    RateLimitMiddleware,
    RateLimitRule,
    key_by_jwt_subject,
    key_by_plan_principal,
    plan_by_jwt_claim,
)

from src.api.dependencies.resources import get_jwt_utils


def create_app() -> FastAPI:
    jwt = get_jwt_utils()
    policy = PlanRateLimitPolicy(
        {
            "free": [
                RateLimitRule(60, 60.0, scope="minuto"),
                RateLimitRule(1_000, 86_400.0, scope="dia"),
            ],
            "pro": [
                RateLimitRule(600, 60.0, burst=100, scope="minuto"),
                RateLimitRule(100_000, 86_400.0, scope="dia"),
            ],
        },
        resolve=plan_by_jwt_claim(jwt, "plan"),
        default_plan="free",
    )
    app = FastAPI()
    app.add_middleware(
        RateLimitMiddleware,
        policy=policy,
        key_func=key_by_plan_principal(policy, key_by_jwt_subject(jwt)),
    )
    return app

Três decisões que valem entender:

• Plano desconhecido cai no default_plan. Um tier que a configuração não conhece precisa ser limitado, não servido sem limite — e uma requisição é o lugar errado para descobrir um erro de digitação. O que é validado na construção: plans vazio, default_plan fora do mapa, e plano sem nenhuma regra. Os três só apareceriam em produção como tráfego ilimitado.
• Rejeição não gasta nada. Uma requisição barrada pelo teto diário não pode queimar token do limite por minuto — senão o minuto drena sem uma única requisição servida. Os dois stores decidem todas as regras antes de escrever qualquer uma; o RedisQuotaStore faz isso dentro de um script Lua, que é a única forma de manter a lista inteira atômica.
• key_by_plan_principal prefixa o plano na chave. Quem sobe de tier passa a escrever em contadores novos em vez de herdar os esgotados. O preço é que quem escolhe o próprio plano (um header vindo do cliente) zeraria o contador só trocando de valor — use com um resolver que a sua borda controla.

O corpo do 429 é o envelope de erro do SDK

O 429 sai no mesmo formato que register_exception_handlers escreve em todo handler, então o cliente parseia um envelope para toda falha:

{
    "detail": "Too many requests",
    "code": "TOO_MANY_REQUESTS",
    "details": {"retry_after_seconds": 60, "limit": 15}
}

code é o campo em que o cliente ramifica — nunca detail, que é prosa e muda com a locale negociada quando existe um MessageCatalog. Troque os dois pelo que a sua API usa:

from fastapi import FastAPI

from tempest_fastapi_sdk import RateLimitMiddleware


app = FastAPI()

app.add_middleware(
    RateLimitMiddleware,
    max_requests=15,
    window_seconds=1.0,
    error_message="Calma parceiro, você está fazendo muitas requisições!",
    error_code="TOO_MANY_REQUESTS",
)

details carrega o que antes só existia em header: retry_after_seconds (o mesmo número do Retry-After) e limit da regra que barrou — no modo policy, a mais apertada.

Mudou na v0.256.0

Até a v0.255.0 o 429 saía como text/plain com o error_message cru no corpo. Quem lê esse corpo como texto precisa passar a ler JSON e pegar detail; quem já ramificava por status === 429 não muda nada.

A troca fecha uma contradição do próprio SDK: error_responses() sempre apontou o 429 para o ErrorResponseSchema, então cliente gerado a partir do OpenAPI quebrava ao desserializar o texto.

Documentando o 429 no OpenAPI

error_responses(TooManyRequestsException) descreve exatamente o que o middleware envia — o error_code default é o code dessa exceção. Se você trocar o error_code, troque também a exceção que documenta a rota, ou o schema volta a divergir do corpo.

Por que o middleware não levanta a exceção

Seria a saída óbvia — levantar TooManyRequestsException e deixar o handler registrado formatar. Não funciona: BaseHTTPMiddleware adicionado por add_middleware fica fora do ExceptionMiddleware do Starlette, então exceção levantada no dispatch não encontra handler e vira 500. Por isso o middleware monta a resposta ele mesmo, lendo o code default da própria exceção para os dois não divergirem.

Headers RateLimit-*

Toda resposta sai com RateLimit-Limit e RateLimit-Remaining (desligue com limit_headers=False), descrevendo a regra mais apertada — a que o cliente deve usar para se auto-regular. RateLimit-Reset só sai quando o número é conhecido: sempre no modo policy, e apenas no 429 no modo janela simples. O store de janela deslizante não reporta reset para requisição aceita, e um número chutado é pior que header ausente.

Multi-réplica

# src/api/app.py

from fastapi import FastAPI
from redis.asyncio import Redis

from tempest_fastapi_sdk import (
    RateLimitMiddleware,
    RateLimitRule,
    RedisQuotaStore,
    StaticRateLimitPolicy,
)

from src.core.settings import settings


def create_app() -> FastAPI:
    redis: Redis = Redis.from_url(settings.REDIS_URL)
    app = FastAPI()
    app.add_middleware(
        RateLimitMiddleware,
        policy=StaticRateLimitPolicy(
            [RateLimitRule(600, 60.0, burst=100)],
        ),
        quota_store=RedisQuotaStore(redis),        # ← compartilhado entre réplicas
    )
    return app

Regra de janela vira sorted set de timestamps; regra de bucket vira um hash com tokens + ts. O TTL da chave de bucket cobre o tempo de encher o balde, não a janela: um bucket lento com rajada grande (10 tokens/minuto, burst=1000) leva mais de uma hora para encher, e expirar antes disso devolveria um balde cheio de graça. Em erro do Redis, fail_open=True (default) libera a requisição.

Relógio: o do Redis, não o da réplica

Recarga de bucket e poda de janela dependem de "quanto tempo passou". Com várias réplicas, a pergunta é: pelo relógio de quem?

Os dois stores de Redis (RedisQuotaStore e RedisRateLimitStore) respondem "pelo do Redis": o script Lua lê TIME do próprio servidor, e nenhuma réplica manda timestamp. Assim, réplicas com relógio de parede dessincronizado — ou uma VM que acabou de dar um salto de NTP — continuam vendo um balde e uma janela.

O que isso evita, medido

Até a v0.301.0 cada réplica mandava o time.time() dela. Contra Redis 4 e 7 reais, com dois clientes — um esgotando o limite, outro sondando:

• um cliente uma hora adiantado passou num bucket (1 token/hora, burst=3) e numa janela (3/hora) que o outro tinha acabado de esgotar;
• um cliente uma hora atrasado gravou timestamps que deixaram o cliente no horário passar do mesmo jeito.

Com o TIME do servidor, os oito casos são negados. Eles ficam em tests/api/test_quota_redis_live.py (make test-docker).

Duas consequências práticas:

• O relógio que importa é o do host do Redis. Um salto nele ainda move todas as chaves de uma vez — mas é um relógio só, em vez de um por réplica.
• Redis 4 funciona, porque o script liga a replicação por efeito (redis.replicate_commands()) antes de escrever. Sem isso o Redis 4.0.14 recusa a escrita depois do TIME, e com fail_open=True um limite de 1 requisição deixou passar 10 de 10.

redis-py 8 e Redis anterior ao 6

O redis-py 8.1.0 — o piso do extra [cache] — abre a conexão com HELLO 3, que o Redis 4 e o 5 recusam como comando desconhecido. Contra esses servidores, crie o client com protocol=2: Redis.from_url(settings.REDIS_URL, protocol=2).

Testando sem depender da velocidade da máquina

O MemoryQuotaStore aceita clock=: qualquer callable que devolve segundos. Com um relógio que só anda quando o teste manda, "o burst se esgota" e "o token volta depois de uma hora" viram asserções exatas — não uma corrida contra a taxa de recarga:

import asyncio

from tempest_fastapi_sdk import MemoryQuotaStore, RateLimitRule


class ManualClock:
    def __init__(self) -> None:
        self.now: float = 0.0

    def __call__(self) -> float:
        return self.now


async def main() -> None:
    clock = ManualClock()
    store = MemoryQuotaStore(clock=clock)
    rules = [RateLimitRule(max_requests=1, window_seconds=3600.0, burst=2)]

    print([(await store.consume("k", rules)).allowed for _ in range(3)])
    clock.now += 3600.0
    print((await store.consume("k", rules)).allowed)


asyncio.run(main())

Saída:

[True, True, False]
True

Por que não sleep?

Uma regra de 100 por segundo repõe um token a cada 10 ms. Quatro chamadas seguidas numa máquina carregada passam disso, o token volta, e o teste "o burst se esgota" falha: foi o que aconteceu com a própria suíte do SDK (#339). Taxa desprezível na janela do teste (1 por hora) ou relógio injetado tiram a velocidade da máquina da asserção.

Rate limit por rota (make_rate_limit_dependency)

O RateLimitMiddleware limita o app inteiro, e o escopo dele só diminui por subtração (exempt_paths). Duas perguntas ficam de fora:

• "Só POST /api/invites." Uma rota cara merece um teto que o resto do app não tem.
• "Por e-mail convidado, não só por IP." A chave está no corpo, e o key_func do middleware recebe só o Request, de forma síncrona, antes de o corpo ser lido.

make_rate_limit_dependency responde as duas: é uma dependency que você pendura em uma rota, conta no mesmo RateLimitStore do middleware (memória ou Redis) e devolve o mesmo 429.

Limitar um endpoint por IP e por campo do corpo

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

from tempest_fastapi_sdk import (
    MemoryRateLimitStore,
    key_by_body_field,
    key_by_ip,
    make_rate_limit_dependency,
    register_exception_handlers,
)


class InviteCreateSchema(BaseModel):
    email: str


invite_limit = make_rate_limit_dependency(
    MemoryRateLimitStore(),
    max_requests=3,
    window_seconds=3600.0,
    key=[
        key_by_ip(trusted_header="x-real-ip"),
        key_by_body_field("email"),
    ],
)

app = FastAPI()
register_exception_handlers(app)


@app.post("/api/invites", status_code=202, dependencies=[Depends(invite_limit)])
async def create_invite(payload: InviteCreateSchema) -> dict[str, str]:
    return {"invited": payload.email}


@app.get("/api/invites")
async def list_invites() -> list[str]:
    return []


client = TestClient(app)
for ip in ("10.0.0.1", "10.0.0.2", "10.0.0.3", "10.0.0.4"):
    response = client.post(
        "/api/invites",
        json={"email": "ana@example.com"},
        headers={"x-real-ip": ip},
    )
    print(response.status_code, response.json())
print(response.headers["retry-after"], response.headers["ratelimit-limit"])
print(client.get("/api/invites").status_code)

Saída:

202 {'invited': 'ana@example.com'}
202 {'invited': 'ana@example.com'}
202 {'invited': 'ana@example.com'}
429 {'detail': 'Too many requests', 'code': 'TOO_MANY_REQUESTS', 'details': {'retry_after_seconds': 3600, 'limit': 3}}
3600 3
200

Quatro IPs diferentes, o mesmo e-mail: o quarto pedido cai pela chave do e-mail, e o GET da mesma URL continua livre. Pedaço por pedaço:

• key=[...] — cada função devolve uma chave (ou várias, ou nenhuma), e cada chave conta no mesmo store. A primeira que estoura recusa o pedido; as seguintes nem são contadas, para um pedido recusado não gastar o orçamento delas. Funções síncronas (key_by_ip, key_by_header, key_by_jwt_*) e assíncronas valem igual.
• key_by_body_field("email") — lê o campo do corpo JSON. O FastAPI lê o corpo antes de resolver as dependencies e o Starlette guarda os bytes no Request, então o endpoint continua recebendo o payload validado — a saída acima é essa prova. O valor é normalizado (strip + minúsculas) e guardado como SHA-256, para o Redis não acumular e-mail em texto puro (normalize=False, hash_value=False desligam). Corpo que não é JSON, ou sem o campo, não gera chave: o 422 do próprio endpoint responde esse pedido.
• trusted_ip_header — sem key, a chave default é o IP resolvido por esse header (make_rate_limit_dependency(store, ..., trusted_ip_header="x-real-ip")). Com key, passe o header para o key_by_ip de dentro da lista; passar os dois levanta ValueError na construção, em vez de ignorar o header calado.
• O balde é da rota. O prefixo default de cada chave é o método e o path template ("POST /api/invites"), então a mesma dependency em duas rotas dá um orçamento para cada, /users/1 e /users/2 dividem o de /users/{user_id}, e o contador nunca colide com o do middleware no mesmo Redis. Passe scope="invites" para várias rotas dividirem um orçamento.

O mesmo 429 do middleware

O corpo (detail, code, details.retry_after_seconds, details.limit) e os headers (Retry-After, RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset) saem dos mesmos helpers que o RateLimitMiddleware usa — tests/api/test_rate_limit_dependency.py compara os dois. A diferença é o caminho: a dependency levanta TooManyRequestsException, então precisa do register_exception_handlers no app, e com um MessageCatalog registrado ali o detail sai traduzido pelo code ("Requisições em excesso" em pt-BR), o que o middleware — que responde fora dos handlers — não faz.

Store criado no create_app

Passe uma função que recebe o Request no lugar do store — make_rate_limit_dependency(store_of, ...), com store_of(request) devolvendo request.app.state.rate_limit_store — e a dependency declarada no import conta no store do app que atende o request. Medido com dois apps e a mesma dependency: cada um conta no seu.

Multi-réplica

Troque MemoryRateLimitStore() por RedisRateLimitStore(redis) e todas as réplicas dividem os contadores, exatamente como no middleware.

Limite de tamanho do body (BodySizeLimitMiddleware)

Um upload de 2 GB num endpoint que espera JSON derruba o worker antes de qualquer validação do Pydantic. BodySizeLimitMiddleware é ASGI puro e corta isso na porta:

# src/api/app.py
from fastapi import FastAPI
from tempest_fastapi_sdk import BodySizeLimitMiddleware

app: FastAPI = FastAPI()

app.add_middleware(
    BodySizeLimitMiddleware,
    max_bytes=2 * 1024 * 1024,             # 2 MiB pra qualquer request
    exclude_paths=("/api/files/upload",),  # rota de upload tem limite próprio
)

São duas checagens:

1. Header — Content-Length acima do teto responde 413 na hora, sem ler um byte do corpo. Pega o caso comum, em que o cliente sabe o tamanho.
2. Streaming — em upload chunked (sem Content-Length), o middleware conta os bytes das mensagens http.request e aborta ao cruzar o teto.

A resposta é o envelope canônico do SDK: {"detail": "Request body too large.", "code": "REQUEST_BODY_TOO_LARGE", "details": {"max_bytes": 1048576}} com HTTP 413.

exclude_paths casa por prefixo

O match é startswith, então quanto mais específico o prefixo, melhor — ("/api/files",) libera tudo abaixo de /api/files. Use pra rota que aceita arquivo grande de propósito e aplica o próprio limite (via UploadUtils(max_size=...), por exemplo). max_bytes=0 desliga a checagem — não suba isso pra produção.

Cache de resposta HTTP (ETag / 304)

ResponseCacheMiddleware entrega dois ganhos de performance em camadas.

ETag + GET condicional (sempre ligado). Toda resposta cacheável ganha um ETag forte (hash do corpo) e um Cache-Control. Quando o cliente manda um If-None-Match que bate, o middleware responde 304 Not Modified sem corpo — o handler ainda roda, mas os bytes não vão pela rede.

from fastapi import FastAPI

from tempest_fastapi_sdk import ResponseCacheMiddleware


def create_app() -> FastAPI:
    app = FastAPI()
    app.add_middleware(ResponseCacheMiddleware, ttl_seconds=30)   # só ETag/304
    return app

Cache server-side (opt-in via store=). Com um store ligado, uma resposta GET/HEAD cacheável é guardada por ttl_seconds; uma requisição igual depois é servida sem rodar o handler (X-Cache: HIT), e um If-None-Match que bate no ETag guardado ainda curto-circuita pro 304.

from typing import Any

from fastapi import FastAPI

from tempest_fastapi_sdk import RedisResponseCacheStore, ResponseCacheMiddleware


def create_app(redis: Any) -> FastAPI:
    app = FastAPI()
    app.add_middleware(
        ResponseCacheMiddleware,
        store=RedisResponseCacheStore(redis),   # compartilhado entre réplicas
        ttl_seconds=60,
        vary=("Accept-Encoding",),               # varia a chave e emite Vary
    )
    return app

Só métodos seguros (GET/HEAD) e respostas de sucesso (200 por padrão) são cacheados. Respostas que optam por fora — Cache-Control: no-store/private ou com Set-Cookie (personalizadas) — nunca são guardadas.

Chave de cache

A chave é método|path|query mais os headers listados em vary=. Passe cacheable=<predicado> pra excluir requisições específicas, ou exempt_paths=(...) pra pular paths exatos. O store espelha o do idempotency (memória ou Redis com client cru), então compõe com o mesmo Redis do serviço.

Stream nunca é drenado (e exempt_paths é igualdade)

Cachear significa ler o corpo até o fim para hashear e guardar. Num text/event-stream o corpo só acaba quando o cliente desconecta, então até a v0.285.0 uma rota SSE atrás deste middleware nunca respondia: a requisição pendurava até o proxy_read_timeout do nginx e virava 504.

Desde a v0.286.0 o middleware passa o stream adiante intacto. Não há o que configurar: uma resposta com Content-Type: text/event-stream, ou sem Content-Length, não é cacheada nem ganha ETag.

O mesmo valia para IdempotencyMiddleware

Ele drena o corpo para poder reproduzir a resposta num retry, e não tinha lista de isenção nenhuma. Um POST que devolve stream — como o /generate/stream do make_genai_router — pendurava do mesmo jeito quando o cliente mandava Idempotency-Key. Também passa adiante agora: um stream não é reproduzível a partir de store nenhum.

exempt_paths mudou de significado em dois middlewares. Ele era igualdade em ResponseCacheMiddleware, RateLimitMiddleware e GracefulShutdownMiddleware, e prefixo em AccessLogMiddleware e HoneypotBanMiddleware — o mesmo nome de argumento, dois comportamentos. Um serviço que isentava sua rota SSE dos dois passava a mesma tupla para os dois e ganhava isenção num e nada no outro.

Agora exempt_paths é igualdade nos cinco, e prefixo tem nome próprio:

from fastapi import FastAPI

from tempest_fastapi_sdk import AccessLogMiddleware, ResponseCacheMiddleware


def create_app() -> FastAPI:
    """Build an app whose SSE subtree is exempt from both middlewares."""
    app: FastAPI = FastAPI()
    app.add_middleware(
        ResponseCacheMiddleware,
        ttl_seconds=30,
        exempt_paths=("/health/liveness",),      # este path, e só ele
        exempt_prefixes=("/api/sse",),           # e tudo sob /api/sse
    )
    app.add_middleware(
        AccessLogMiddleware,
        exempt_prefixes=("/api/sse",),           # antes: exempt_paths
    )
    return app

Migração: exempt_paths de prefixo vira exempt_prefixes

Só AccessLogMiddleware e HoneypotBanMiddleware mudam de comportamento. Se você passava exempt_paths=("/api/sse",) a um deles contando que cobrisse /api/sse/stream, renomeie para exempt_prefixes — caso contrário a isenção deixa de valer para as rotas filhas.

O alargamento contrário — tornar tudo prefixo — foi recusado de propósito: em RateLimitMiddleware uma isenção é um buraco no rate limit, e exempt_paths=("/health",) viraria isenção de /health-admin também, sem ninguém pedir.

Requisição autenticada não divide entrada de cache

Repare no que a chave não tem: quem pediu. Se GET /api/me de duas pessoas cai na mesma chave, a primeira resposta guardada é servida pra segunda — e pra qualquer anônimo que peça o mesmo path.

Por isso uma requisição que carrega Authorization ou Cookie passa por fora do store compartilhado. Ela continua ganhando ETag e 304 (que são por resposta, e seguros); só não entra num cache que outra pessoa pode ler.

from typing import Any

from fastapi import FastAPI

from tempest_fastapi_sdk import RedisResponseCacheStore, ResponseCacheMiddleware


def create_app(redis: Any) -> FastAPI:
    app = FastAPI()
    app.add_middleware(
        ResponseCacheMiddleware,
        store=RedisResponseCacheStore(redis),
        ttl_seconds=60,
        cache_credentialed=True,   # opt-in: chave passa a incluir a credencial
    )
    return app

Com cache_credentialed=True um digest dos headers de credencial entra na chave, então cada chamador ganha a sua entrada e o cache volta a valer pra rota autenticada. Ligue só depois de conferir que as rotas cacheadas não carregam uma segunda identidade que a credencial não representa (um header de tenant, uma sessão independente do path) — nesse caso a credencial sozinha não separa o suficiente.

Cache-Control padrão é private

O default emitido é private, max-age=<max_age>. O middleware não tem como saber se o corpo de uma rota é personalizado, e um public errado ali faz um proxy compartilhado servir a resposta de um usuário pra outro. Passe cache_control="public, max-age=…" de propósito, num router que só serve conteúdo compartilhado.

Verificação de assinatura de webhook

WebhookSignatureVerifier valida webhooks de entrada assinados com HMAC (estilo GitHub) e expõe uma dependência FastAPI que lê o corpo cru, checa a assinatura com hmac.compare_digest e entrega os bytes do corpo para que o handler da rota possa reparsear sem reler o stream.

# src/api/dependencies/webhooks.py
from tempest_fastapi_sdk import WebhookSignatureVerifier

from src.core.settings import settings


github = WebhookSignatureVerifier(
    secret=settings.GITHUB_WEBHOOK_SECRET,
    algorithm="sha256",
    header_name="X-Hub-Signature-256",
    prefix="sha256=",
)

Stripe não cabe neste verificador

O Stripe-Signature não é um hex do HMAC do corpo: é t=<timestamp>,v1=<hex>, e o que a Stripe assina é "<t>." + corpo. Um WebhookSignatureVerifier(header_name="Stripe-Signature", encoding="hex") recusa toda entrega legítima — medido com um header gerado por sign_payload, verify devolve False. Para a Stripe use make_stripe_webhook_dependency da integração Stripe, que parseia o header e checa a janela de tempo.

# src/api/routers/webhooks.py

import json

from fastapi import APIRouter, Depends

from src.api.dependencies.webhooks import github


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


@router.post("/github")
async def github_event(body: bytes = Depends(github.dependency())) -> None:
    payload = json.loads(body)
    ...

Suporta encodings hex (default) e base64, qualquer algoritmo hashlib garantido entre plataformas, e um prefix opcional (ex.: "sha256=") removido antes da comparação. Use o imperativo verifier.verify(body, signature) de handlers de fila quando a validação acontece fora do pipeline FastAPI.

A assinatura cobre só o corpo, então uma entrega capturada continua válida pra sempre — quem observar uma pode reenviar indefinidamente. Passe timestamp_header= pra também exigir um timestamp unix recente e recusar o que estiver fora da janela:

from tempest_fastapi_sdk import WebhookSignatureVerifier

from src.core.settings import settings


github = WebhookSignatureVerifier(
    settings.GITHUB_WEBHOOK_SECRET,
    header_name="X-Hub-Signature-256",
    prefix="sha256=",
)

check_github = github.dependency(
    timestamp_header="X-Webhook-Timestamp",
    max_age_seconds=300,
)

É opt-in porque provedor que não manda esse header teria o tráfego legítimo recusado. O WebhookSender do próprio SDK envia X-Webhook-Timestamp.

O timestamp não entra na assinatura

Como só o corpo é assinado, quem replica uma captura também pode reescrever o timestamp pra um atual. A checagem limita replay acidental e oportunista, não um atacante ativo que lê e reescreve a request. Fechar isso depende do provedor assinar timestamp.body — o formato que a Stripe usa (e que a integração Stripe já verifica); quando o seu provedor fizer isso, valide com verifier.verify(f"{ts}.".encode() + body, signature).

Para provedores que assinam com uma chave privada RSA (Apple App Store, Google Play, serviços enterprise custom), troque WebhookSignatureVerifier por RSAWebhookSignatureVerifier — mesma superfície verify(body, signature), mas valida a assinatura contra uma chave pública codificada em PEM. Usa RSASSA-PKCS1-v1_5 sobre SHA-256/384/512 (configurável via algorithm=). Requer o pacote cryptography (instalado com o extra [webpush]).

from tempest_fastapi_sdk import RSAWebhookSignatureVerifier

from src.core.settings import settings

base64_signature_header_value = "c2lnbmF0dXJl"
raw_body_bytes = b'{"event": "order.paid"}'


apple = RSAWebhookSignatureVerifier(
    public_key_pem=settings.APPLE_PUBLIC_KEY_PEM,
    header_name="X-Apple-Signature",
    algorithm="sha256",
)

# Em handlers de fila / fora do FastAPI:
ok: bool = apple.verify(raw_body_bytes, base64_signature_header_value)

Entrega de webhooks de saída — WebhookSender

A contraparte: enviar eventos assinados pros seus assinantes. WebhookSender faz POST do evento em JSON, assina o corpo com o mesmo WebhookSignatureVerifier (então o receptor valida com aquele verifier) e re-tenta falhas transitórias (erro de conexão, 5xx, 429) com backoff exponencial. Outros 4xx não são re-tentados. O cliente httpx é injetado (você é dono do ciclo de vida).

Instalação

O WebhookSender usa httpx, que já vem no pacote base tempest-fastapi-sdk — nenhum extra é necessário. O extra [http] continua existindo e é opcional: declara o httpx de forma explícita.

import asyncio

import httpx

from tempest_fastapi_sdk import WebhookSender, WebhookSignatureVerifier

from src.core.settings import settings

order = {"id": "abc", "status": "paid"}
subscribers = ["https://partner.example.com/hooks"]


verifier = WebhookSignatureVerifier(settings.WEBHOOK_SECRET, prefix="sha256=")


async def main() -> None:
    """Run this example."""
    async with httpx.AsyncClient() as client:
        sender = WebhookSender(client, signer=verifier, max_attempts=4)
        result = await sender.send(
            "https://assinante.example.com/hooks",
            event="order.paid",
            payload={"id": order["id"], "total": 4200},
        )
        if not result.delivered:
            # result.status_code / result.attempts / result.error
            ...  # enfileira pra reprocessar, alerta, etc.

    # Mesmo evento pra vários assinantes, concorrente:
    results = await sender.send_many(
        [(url, {"id": order["id"]}) for url in subscribers],
        event="order.paid",
    )


asyncio.run(main())

Cada entrega envia os headers X-Webhook-Event, X-Webhook-Id (uuid único) e X-Webhook-Timestamp, mais a assinatura HMAC no header do signer. Devolve um WebhookDelivery (delivered, status_code, attempts, error, delivery_id).

Casa com o outbox

Pareie com BaseOutboxModel + OutboxRelay: grave o evento na mesma transação do negócio e deixe o relay chamar o WebhookSender — entrega ao menos uma vez, com a assinatura que o assinante verifica.

Headers Link de paginação

build_pagination_link_header emite um header Link RFC 8288 com os rels first / prev / next / last — combine-o com (ou use no lugar de) o wrapper de corpo BasePaginationSchema para clientes REST que esperam headers no estilo GitHub. Os query parameters existentes na URL base são preservados.

from fastapi import APIRouter, Depends, Request, Response

from tempest_fastapi_sdk import BasePaginationSchema, build_pagination_link_header

from src.api.dependencies.controllers import get_user_controller
from src.controllers import UserController
from src.schemas import UserFilterSchema, UserResponseSchema

router = APIRouter()


@router.get("", response_model=list[UserResponseSchema])
async def list_users(
    request: Request,
    response: Response,
    filters: UserFilterSchema = Depends(),
    controller: UserController = Depends(get_user_controller),
) -> list[UserResponseSchema]:
    result = await controller.paginate(
        filters=filters.get_conditions(),
        order_by=filters.order_by,
        page=filters.page,
        page_size=filters.page_size,
        ascending=filters.ascending,
    )
    page = BasePaginationSchema[UserResponseSchema](**result)
    response.headers["Link"] = build_pagination_link_header(
        str(request.url),
        page=page.page,
        page_size=page.page_size,
        pages=page.pages,
    )
    response.headers["X-Total-Count"] = str(page.total)
    return page.items

Ajuste page_param= / size_param= quando seu serviço usa nomes de query parameter não-padrão (ex.: offset / limit). Passe extra_params={"sort": "name"} para embutir o estado atual de sort/filtro em cada link.

Router de tool-spec

make_tool_spec_router(spec) monta um endpoint GET /tool-spec expondo um manifesto legível por máquina no prefixo raiz — pensado para ficar ao lado de /health/liveness para que callers externos possam descobrir capacidades sem parsear o documento OpenAPI completo.

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import make_health_router, make_tool_spec_router

from src.api.dependencies.resources import db
from src.core.settings import settings


def _tool_spec() -> dict[str, object]:
    """Computed per request — keeps version + counts in sync with state."""
    return {
        "service": "my-service",
        "version": settings.VERSION,
        "tools": [
            {"path": "/api/users", "method": "GET", "summary": "List users"},
            {"path": "/api/orders", "method": "POST", "summary": "Place order"},
        ],
    }


def create_app() -> FastAPI:
    app = FastAPI(...)
    app.include_router(make_health_router(db=db))
    app.include_router(make_tool_spec_router(_tool_spec))
    ...
    return app

Passe um dict (servido literalmente), um callable sync (chamado a cada requisição) ou um callable async (aguardado). Sobrescreva path= para expor o manifesto em uma URL diferente ou tag= para agrupá-lo sob uma tag OpenAPI diferente.

Ponto de entrada programático do servidor

run_server é o helper canônico importado de src/server.py. Ele centraliza os defaults de host / port / reload — puxando valores de um objeto settings no estilo ServerSettings quando presente — e mantém o ponto de entrada em uma única linha.

# src/server.py
from tempest_fastapi_sdk import run_server

from src.api.app import app  # noqa: F401 — re-exported for external runners
from src.core.settings import settings


def run() -> None:
    """Start the API server programmatically."""
    run_server("src.api.app:app", settings=settings)


__all__: list[str] = ["app", "run"]

# main.py
from src.server import run

if __name__ == "__main__":
    run()

A ordem de resolução de cada kwarg é argumento explícito → settings.SERVER_* → default do SDK ("127.0.0.1" / 8000 / False). Kwargs extras do uvicorn (workers=, log_config=, ssl_*=) são encaminhados literalmente.

Composição de mixins de settings

BaseAppSettings é a base pydantic-settings configurada (env_file=".env", extra="ignore", case_sensitive=True). O SDK também expõe mixins componíveis para as dependências mais comuns; escolha os que o serviço precisa e componha na sua classe Settings.

BaseAppSettings deve ser a última base

Cada mixin herda BaseAppSettings, então todos carregam o mesmo model_config — inclusive env_file=".env" — e o .env é lido independente da ordem em que os mixins aparecem. Mas BaseAppSettings deve ser a última base: como os mixins são subclasses dela, listá-la antes de qualquer mixin viola a linearização C3 do Python e quebra o import. A partir da v0.159.1 a metaclasse AppSettingsMeta intercepta isso e levanta TypeError: Settings: BaseAppSettings must be the LAST base — DatabaseSettings already subclasses it ..., que já indica a correção; antes dela a mensagem era o TypeError: Cannot create a consistent method resolution order (MRO) cru do pydantic. Ponha sempre BaseAppSettings no fim das bases — veja o guia de migração.

# src/core/settings.py
from pydantic import Field

from tempest_fastapi_sdk import (
    BaseAppSettings,
    CORSSettings,
    DatabaseSettings,
    EmailSettings,
    JWTSettings,
    LogSettings,
    RabbitMQSettings,
    RedisSettings,
    ServerSettings,
    TaskIQSettings,
    TokenSettings,
    UploadSettings,
    WebPushSettings,
)


class Settings(
    ServerSettings,
    LogSettings,
    DatabaseSettings,
    RedisSettings,
    RabbitMQSettings,
    TaskIQSettings,
    JWTSettings,
    CORSSettings,
    EmailSettings,
    UploadSettings,
    TokenSettings,
    WebPushSettings,
    BaseAppSettings,
):
    """Service-wide settings."""

    VERSION: str = Field(default="0.0.0")


settings = Settings()

Cada mixin é dono do seu próprio prefixo de env var — escolha só os que o serviço precisa:

Mixin	Env vars
ServerSettings	SERVER_HOST, SERVER_PORT, SERVER_RELOAD, SERVER_DEBUG
LogSettings	LOG_LEVEL, LOG_JSON, LOG_DIR, LOG_MAX_BYTES, LOG_BACKUP_COUNT, LOG_REDACT
DatabaseSettings	DATABASE_URL, DATABASE_ECHO, DATABASE_POOL_SIZE, DATABASE_MAX_OVERFLOW, DATABASE_POOL_RECYCLE, DATABASE_SQLITE_WAL, DATABASE_SQLITE_BUSY_TIMEOUT, DATABASE_SQLITE_FOREIGN_KEYS
RedisSettings	REDIS_URL, REDIS_DECODE_RESPONSES
RabbitMQSettings	RABBITMQ_URL, RABBITMQ_PREFETCH_COUNT
TaskIQSettings	TASKIQ_BROKER_URL, TASKIQ_RESULT_BACKEND_URL, TASKIQ_STORE_RESULTS, TASKIQ_RESULT_TTL_SECONDS
JWTSettings	JWT_SECRET, JWT_ALGORITHM, JWT_ACCESS_TTL_SECONDS, JWT_REFRESH_TTL_SECONDS, JWT_ISSUER
CORSSettings	CORS_ORIGINS, CORS_ORIGIN_REGEX, CORS_ALLOW_CREDENTIALS, CORS_ALLOW_METHODS, CORS_ALLOW_HEADERS, CORS_EXPOSE_HEADERS, CORS_MAX_AGE
EmailSettings	SMTP_HOST, SMTP_PORT, SMTP_USERNAME, SMTP_PASSWORD, SMTP_FROM_ADDR, SMTP_USE_TLS, SMTP_USE_SSL, SMTP_TIMEOUT_SECONDS
UploadSettings	UPLOAD_DIR, UPLOAD_MAX_SIZE_BYTES, UPLOAD_ALLOWED_EXTENSIONS, UPLOAD_ALLOWED_MIMETYPES
TokenSettings	TOKEN_SECRET
WebPushSettings	VAPID_PUBLIC_KEY, VAPID_PRIVATE_KEY, VAPID_SUBJECT, WEBPUSH_DEFAULT_TTL_SECONDS

Mudança que quebra na 0.8.0: ServerSettings antes expunha os campos crus HOST / PORT / DEBUG / LOG_LEVEL / LOG_JSON. Eles foram renomeados para SERVER_HOST / SERVER_PORT / SERVER_RELOAD / SERVER_DEBUG, e LOG_LEVEL / LOG_JSON migraram para o novo mixin LogSettings. Atualize tanto o seu arquivo .env (nomes de env var) quanto qualquer código lendo settings.HOST etc.

Autenticação

Signup + login + rota protegida de ponta a ponta usando PasswordUtils e JWTUtils. Requer o extra [auth].

Conecte os singletons utilitários

# src/core/security.py
from datetime import timedelta

from tempest_fastapi_sdk import JWTUtils, PasswordUtils

from src.core.settings import settings


passwords = PasswordUtils(rounds=12)

tokens = JWTUtils(
    secret=settings.JWT_SECRET,
    algorithm=settings.JWT_ALGORITHM,
    default_ttl=timedelta(seconds=settings.JWT_ACCESS_TTL_SECONDS),
    issuer="my-app",
)

Signup

Reutilize o UserService.create definido no tutorial — ele já faz hash da senha.

Login

# src/schemas/auth.py
from pydantic import EmailStr

from tempest_fastapi_sdk import BaseSchema


class LoginSchema(BaseSchema):
    email: EmailStr
    password: str


class TokenResponseSchema(BaseSchema):
    access_token: str
    token_type: str = "bearer"

# src/services/auth.py
from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import JWTUtils, PasswordUtils, UnauthorizedException

from src.db.repositories import UserRepository
from src.schemas.auth import LoginSchema, TokenResponseSchema


class AuthService:
    def __init__(
        self,
        session: AsyncSession,
        passwords: PasswordUtils,
        tokens: JWTUtils,
    ) -> None:
        self.repo = UserRepository(session)
        self.passwords = passwords
        self.tokens = tokens

    async def login(self, data: LoginSchema) -> TokenResponseSchema:
        user = await self.repo.get_or_none({"email": data.email})
        if user is None or not self.passwords.verify(
            data.password, user.password_hash
        ):
            # Same error for both cases — don't leak which one failed.
            raise UnauthorizedException(message="E-mail ou senha inválidos")
        token = self.tokens.encode({"sub": str(user.id)})
        return TokenResponseSchema(access_token=token)

# src/api/dependencies/services.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.app import db
from src.core.security import passwords, tokens
from src.services.auth import AuthService


def get_auth_service(
    session: AsyncSession = Depends(db.session_dependency),
) -> AuthService:
    return AuthService(session, passwords, tokens)

# src/api/routers/auth.py
from fastapi import APIRouter, Depends

from src.api.dependencies.services import get_auth_service
from src.schemas.auth import LoginSchema, TokenResponseSchema
from src.services.auth import AuthService


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


@router.post("/login", response_model=TokenResponseSchema)
async def login(
    data: LoginSchema,
    service: AuthService = Depends(get_auth_service),
) -> TokenResponseSchema:
    return await service.login(data)

Proteja uma rota — dependência JWT

Use make_jwt_user_dependency para conectar o esquema bearer + decode do JWT + carga do usuário em uma chamada. A única costura é o user_loader, um callable async que mapeia o claim de subject do JWT para o seu UserModel de domínio — chamado como user_loader(subject) por default, ou user_loader(subject, session) quando você passa session_dependency= (recomendado: assim o usuário volta attached na sessão do request e pode ser mutado sem InvalidRequestError).

O token é procurado em header → cookie → query string, primeiro hit ganha: Authorization: Bearer sempre, mais o cookie de cookie_name= e o parâmetro de query_param= quando você os habilita (ambos None por default). subject_claim= troca o claim lido (default "sub") e error_message= o texto do 401.

Devolver None do user_loader recusa a request (v0.252.0)

O loader é o lugar natural para recusar um usuário — conta desativada, subject que não existe mais, id malformado. Com soft=False (default), None vira 401, a mesma resposta que o subject ausente já dava.

Até a v0.251.0 esse None chegava ao handler e a rota respondia 200 com um usuário que o loader tinha recusado — então desativar uma conta não tinha efeito nenhum até o access token expirar. Com soft=True o None continua chegando: é o propósito do flag.

Levantar UnauthorizedException ou NotFoundException de dentro do loader continua funcionando, e é o que você quer quando a mensagem importa.

# src/api/dependencies/auth.py
from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession
from tempest_fastapi_sdk import make_jwt_user_dependency

from src.api.app import db
from src.core.security import tokens
from src.db.models import UserModel
from src.db.repositories import UserRepository


async def load_user(subject: str, session: AsyncSession) -> UserModel:
    """Resolve the JWT subject (a UUID string) to a persisted user.

    Receives the request-scoped session because the dependency below
    declares `session_dependency=`, so the user comes back attached to
    the same session the route's repositories use. SDK exceptions raised
    inside translate to the canonical 401/404 envelope.
    """
    repo = UserRepository(session)
    return await repo.get_by_id(UUID(subject))


get_current_user = make_jwt_user_dependency(
    tokens,
    load_user,
    session_dependency=db.session_dependency,
)
get_current_user_or_none = make_jwt_user_dependency(
    tokens,
    load_user,
    soft=True,
    session_dependency=db.session_dependency,
)

# Use in any route

from fastapi import APIRouter, Depends

from src.api.dependencies.auth import get_current_user
from src.db.models import UserModel
from src.schemas import UserResponseSchema

router = APIRouter()


@router.get("/me", response_model=UserResponseSchema)
async def me(current: UserModel = Depends(get_current_user)) -> UserResponseSchema:
    return UserResponseSchema.model_validate(current)

Auth suave (usuário opcional)

get_current_user_or_none acima já usa soft=True — ele retorna None em vez de levantar em um token ausente ou inválido, para que endpoints funcionem tanto autenticados quanto anônimos:

from fastapi import APIRouter, Depends

from src.api.dependencies.auth import get_current_user_or_none
from src.db.models import UserModel
from src.schemas import FeedResponseSchema
from src.services import FeedService

feed_service = FeedService()
router = APIRouter()


@router.get("/feed")
async def feed(
    current: UserModel | None = Depends(get_current_user_or_none),
) -> FeedResponseSchema:
    return await feed_service.list(viewer=current)

Por baixo dos panos, soft=True chama tokens.decode_or_none (sem exceção em tokens expirados/inválidos) e pula o loader quando o subject está ausente.

---

Upload de arquivos

Endpoint de avatar com validação + limpeza. Requer o extra [upload].

# src/core/storage.py
from tempest_fastapi_sdk import UploadUtils

from src.core.settings import settings


avatar_storage = UploadUtils(
    f"{settings.UPLOAD_DIR}/avatars",
    max_size_bytes=5 * 1024 * 1024,            # 5 MiB
    allowed_extensions={"png", "jpg", "jpeg", "webp"},
    allowed_mimetypes={"image/png", "image/jpeg", "image/webp"},
    verify_magic_bytes=True,                   # sniff bytes, reject polyglots
)

verify_magic_bytes=True lê os primeiros bytes de cada upload e confirma que o arquivo realmente é um dos tipos permitidos — um payload HTML+JS enviado como image/png é rejeitado mesmo que sua extensão e header Content-Type pareçam válidos. Só ative quando todo formato aceito é um que o sniff_mime reconhece (JPEG, PNG, GIF, BMP, WebP, PDF); caso contrário, um upload legítimo mas não-snifável seria recusado. Para controle mais fino, passe um predicado content_validator para save() (save(file, content_validator=lambda b: sniff_mime(b) in {"image/png"})), e passe filename="..." para um nome determinístico e endereçável (ex.: f"{user_id}.jpg") em vez do UUID padrão.

# src/api/routers/users.py (extension)

from uuid import UUID

from fastapi import APIRouter, Depends, UploadFile

from tempest_fastapi_sdk import ForbiddenException

from src.api.dependencies import get_user_controller
from src.api.dependencies.auth import get_current_user
from src.controllers.user import UserController
from src.core.storage import avatar_storage
from src.db.models import UserModel
from src.schemas import UserResponseSchema

router = APIRouter()


@router.post("/{user_id}/avatar", response_model=UserResponseSchema)
async def upload_avatar(
    user_id: UUID,
    file: UploadFile,
    current: UserModel = Depends(get_current_user),
    controller: UserController = Depends(get_user_controller),
) -> UserResponseSchema:
    if current.id != user_id:
        raise ForbiddenException(message="Só pode editar o próprio avatar")
    path = await avatar_storage.save(file, subdir=str(user_id))
    return await controller.set_avatar(user_id, str(path))

Adicione set_avatar tanto ao service quanto ao controller (o controller fica como um pass-through fino a menos que orquestração seja necessária — ex.: disparar um evento de "avatar atualizado"):

# src/services/user.py

from uuid import UUID

from tempest_fastapi_sdk import UploadUtils

from src.schemas import UserResponseSchema

avatar_storage = UploadUtils(source="./uploads/avatars")


class UserService:
    async def set_avatar(self, user_id: UUID, path: str) -> UserResponseSchema:
        user = await self.repo.get_by_id(user_id)
        # Delete previous file when replacing.
        if user.avatar_path and user.avatar_path != path:
            await avatar_storage.delete(user.avatar_path)
        user.avatar_path = path
        user = await self.repo.update(user)
        return self.repo.map_to_response(user)


# src/controllers/user.py
class UserController:
    async def set_avatar(self, user_id: UUID, path: str) -> UserResponseSchema:
        return await self.service.set_avatar(user_id, path)

UploadUtils.save() levanta FileTooLargeException (413) ou InvalidFileTypeException (415) na rejeição — o exception handler do SDK já retorna o status code certo com um campo code na resposta.

Servindo o arquivo de volta

Uploads em disco local são melhor servidos por um upstream (nginx / Caddy) para que o FastAPI não fique transmitindo bytes. Para dev:

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from src.core.settings import settings

app = FastAPI()


app.mount(
    "/static/uploads",
    StaticFiles(directory=settings.UPLOAD_DIR),
    name="uploads",
)

Construa a URL pública no schema de resposta:

from pydantic import EmailStr, field_validator

from tempest_fastapi_sdk import BaseResponseSchema


class UserResponseSchema(BaseResponseSchema):
    name: str
    email: EmailStr
    avatar_url: str | None = None

    @field_validator("avatar_url", mode="before")
    @classmethod
    def _absolute_url(cls, value: str | None) -> str | None:
        if value is None:
            return None
        # avatar_path stored as relative path → public URL
        return f"/static/uploads/{value}"

Servindo arquivos privados pela API (DownloadUtils)

Quando um arquivo deve ficar atrás de auth — faturas, contratos, exames médicos — uma URL pública /static o vaza para qualquer um que descubra o caminho. DownloadUtils transmite os bytes pelo próprio endpoint, para que os mesmos Depends(get_current_user) / checks de permissão que guardam toda outra rota guardem o download também. Nenhum link público é exposto. Não precisa de nenhum extra (usa FileResponse / StreamingResponse do Starlette, que vêm com o FastAPI).

# src/core/storage.py
from tempest_fastapi_sdk import DownloadUtils

from src.core.settings import settings


invoice_files = DownloadUtils(f"{settings.UPLOAD_DIR}/invoices")

# src/api/routers/invoices.py

from uuid import UUID

from fastapi import APIRouter, Depends
from fastapi.responses import FileResponse

from tempest_fastapi_sdk import ForbiddenException

from src.api.dependencies import get_invoice_controller
from src.api.dependencies.auth import get_current_user
from src.controllers.invoice import InvoiceController
from src.core.storage import invoice_files
from src.db.models import UserModel

router = APIRouter()


@router.get("/{invoice_id}/file")
async def download_invoice(
    invoice_id: UUID,
    current: UserModel = Depends(get_current_user),
    controller: InvoiceController = Depends(get_invoice_controller),
) -> FileResponse:
    invoice = await controller.get_by_id(invoice_id)
    if invoice.owner_id != current.id:
        raise ForbiddenException(message="Fatura de outro usuário")
    # base_dir confines the read — a stored "../../etc/passwd" path 404s.
    return invoice_files.file_response(
        invoice.file_path,                 # relative to base_dir
        filename=f"fatura-{invoice.number}.pdf",
        as_attachment=True,                # force a download dialog
    )

Qualquer caminho relativo que escape de base_dir (traversal ../, caminhos absolutos, escapes via symlink) levanta NotFoundException (404) em vez de vazar o arquivo — o mesmo 404 que você ganha para um arquivo genuinamente ausente, então callers nunca distinguem "proibido" de "ausente". file_response adivinha o tipo MIME pelo nome do arquivo (sobrescreva com media_type=), e as_attachment=False pede inline (ex.: pré-visualizar um PDF no navegador) — atendido só para tipo em INLINE_SAFE_MEDIA_TYPES, então HTML e SVG continuam baixando. Toda resposta de download sai com X-Content-Type-Options: nosniff, CSP sandbox e Cross-Origin-Resource-Policy: same-site.

Para payloads construídos na hora — um relatório gerado, um zip em memória, bytes descriptografados — use stream() em vez de tocar o disco:

import io
from uuid import UUID

from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse

from src.api.dependencies.auth import get_current_user
from src.api.dependencies.controllers import get_invoice_controller
from src.controllers import InvoiceController
from src.core.storage import invoice_files
from src.db.models import UserModel

router = APIRouter()


@router.get("/{invoice_id}/receipt.csv")
async def download_receipt(
    invoice_id: UUID,
    current: UserModel = Depends(get_current_user),
    controller: InvoiceController = Depends(get_invoice_controller),
) -> StreamingResponse:
    csv_bytes: bytes = await controller.render_receipt_csv(invoice_id, current.id)
    return invoice_files.stream(
        csv_bytes,                         # bytes, or a (sync/async) byte generator
        filename="recibo.csv",
    )

stream() aceita bytes cru, um Iterable[bytes] sync ou um AsyncIterable[bytes], então um export grande pode ser entregue pedaço a pedaço sem bufferizar tudo em memória. Ambos os métodos definem um Content-Disposition seguro em UTF-8 (nomes de arquivo não-ASCII sobrevivem via o parâmetro filename* da RFC 5987); build_content_disposition() é exportado se você precisar definir esse header em uma resposta feita à mão.

---

E-mail transacional

Fluxo de reset de senha usando EmailUtils + um JWT de vida curta. Requer o extra [email].

# src/core/mailer.py
from tempest_fastapi_sdk import EmailUtils

from src.core.settings import settings


mailer = EmailUtils(
    host=settings.SMTP_HOST,
    port=settings.SMTP_PORT,
    from_addr=settings.SMTP_FROM_ADDR,
    username=settings.SMTP_USERNAME,
    password=settings.SMTP_PASSWORD,
    use_starttls=True,
)

# src/services/password_reset.py
from datetime import timedelta
from uuid import UUID

from tempest_fastapi_sdk import (
    EmailUtils,
    InvalidTokenException,
    JWTUtils,
    PasswordUtils,
)

from src.db.repositories import UserRepository


class PasswordResetService:
    def __init__(
        self,
        repo: UserRepository,
        tokens: JWTUtils,
        mailer: EmailUtils,
    ) -> None:
        self.repo = repo
        self.tokens = tokens
        self.mailer = mailer

    async def request_reset(self, email: str) -> None:
        """Send a password-reset link to `email`.

        Always returns silently — don't reveal whether the email
        is registered or not (avoids account enumeration).
        """
        user = await self.repo.get_or_none({"email": email})
        if user is None:
            return
        token = self.tokens.encode(
            {"sub": str(user.id), "purpose": "password_reset"},
            ttl=timedelta(minutes=15),
        )
        reset_url = f"https://my-app.com/reset-password?token={token}"
        await self.mailer.send(
            to=user.email,
            subject="Reset your password",
            body=f"Click here to reset your password: {reset_url}",
            html=f'<p>Click <a href="{reset_url}">here</a> to reset.</p>',
        )

    async def consume_reset(
        self,
        token: str,
        new_password: str,
        passwords: PasswordUtils,
    ) -> None:
        # `decode` raises InvalidTokenException / ExpiredTokenException
        # (both 401). Caught by the SDK handler.
        payload = self.tokens.decode(token)
        if payload.get("purpose") != "password_reset":
            raise InvalidTokenException()
        user = await self.repo.get_by_id(UUID(payload["sub"]))
        user.password_hash = passwords.hash(new_password)
        await self.repo.update(user)

---

Recap / próximos passos

Você agora conhece a superfície HTTP inteira: bootstrap do app, exception handlers com i18n, dependências de auth, rate limit, verificação de webhook, headers de paginação, tool-spec, o ponto de entrada do servidor e a composição de settings. As três últimas seções (autenticação, upload, e-mail) foram um resumo — cada uma tem uma receita dedicada que vai mais fundo:

• Fluxo de autenticação » — o flow bundled completo (UserAuthService + make_auth_router): signup, ativação, login, reset de senha, refresh tokens e current_user.
• Upload de arquivos » — backends de storage plugáveis (LocalUploadStorage / MinIOUploadStorage), URLs pré-assinadas e validação de conteúdo.
• E-mail transacional » — EmailUtils com templates Jinja2, os defaults embutidos (activation.html, password_reset.html) e como sobrescrevê-los.
• Downloads privados » — servir arquivos atrás de auth com DownloadUtils (file_response / stream), sem vazar links públicos.