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.
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:
- 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. - 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.

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.

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 na construção — 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.

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:

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.
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.

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 Stripe / 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 = WebhookSignatureVerifier(
    secret=settings.STRIPE_WEBHOOK_SECRET,
    algorithm="sha256",
    header_name="Stripe-Signature",
    encoding="hex",
)

# 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 o Stripe usa; quando o seu 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 resto da camada HTTP já vem com tempest-fastapi-sdk. O WebhookSender depende do extra [http] — uv add "tempest-fastapi-sdk[http]" (traz httpx).

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`
`DatabaseSettings`	`DATABASE_URL`, `DATABASE_ECHO`, `DATABASE_POOL_SIZE`, `DATABASE_MAX_OVERFLOW`, `DATABASE_POOL_RECYCLE`
`RedisSettings`	`REDIS_URL`, `REDIS_DECODE_RESPONSES`
`RabbitMQSettings`	`RABBITMQ_URL`, `RABBITMQ_PREFETCH_COUNT`
`TaskIQSettings`	`TASKIQ_BROKER_URL`, `TASKIQ_RESULT_BACKEND_URL`
`JWTSettings`	`JWT_SECRET`, `JWT_ALGORITHM`, `JWT_ACCESS_TTL_SECONDS`, `JWT_REFRESH_TTL_SECONDS`, `JWT_ISSUER`
`CORSSettings`	`CORS_ORIGINS`, `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",
)

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

# 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/routers/auth.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

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


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


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


@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 serve inline (ex.: pré-visualizar um PDF no navegador).

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.