HTTP layer

Middlewares, dependencies, routers and middleware composition for the API surface.

Here you assemble a service's entire HTTP surface from the primitives tempest_fastapi_sdk ships — without hand-writing middleware, exception handlers or bootstrap glue. Each section is independent: grab only the one you need right now. This recipe covers ~12 primitives:

• create_app() + register_exception_handlers — the canonical bootstrap and the standardized error envelope (with optional i18n via MessageCatalog).
• RequestIDMiddleware — X-Request-ID correlation on every log line.
• apply_cors — CORS from CORSSettings.
• make_health_router / make_token_dependency — liveness/readiness + the X-Token shared-secret guard.
• JWT / bearer / role / permission dependencies — gate routes by token and by role.
• RateLimitMiddleware — sliding window, key by IP/user/tenant, memory or Redis store.
• BodySizeLimitMiddleware — hard cap on the request body, answering 413 before any parsing.
• WebhookSignatureVerifier / RSAWebhookSignatureVerifier — validate signed webhooks (HMAC or RSA).
• build_pagination_link_header — GitHub-style RFC 8288 Link header.
• make_tool_spec_router — machine-readable manifest at the root prefix.
• run_server — programmatic uvicorn entry point.
• BaseAppSettings + *Settings mixins — composable env-var configuration.

The last three sections are full flows

Authentication, uploads and transactional email appear here in condensed form; each has a dedicated, deeper recipe — see the Recap at the end of the page.

Application bootstrap

Section 2 of the tutorial shows the minimal create_app(). This recipe is the extended version, wiring everything tempest_fastapi_sdk.api ships — exception handlers, CORS, request-ID middleware, the health router with extra checks, a shared-secret token dependency and an extra Redis manager — all from the same canonical src/api/app.py location. The bootstrapping pattern stays identical; only the contents of create_app() grow.

# 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()

Key points:

• src/server.py and main.py (one-liner) stay exactly as in Section 2 of the tutorial — only create_app() changes when you add primitives. Never start uvicorn via subprocess.run(["uvicorn", ...]); always import app from src.api.app or call uvicorn.run("src.api.app:app", ...) programmatically from src/server.py.
• RequestIDMiddleware reads/writes X-Request-ID and seeds request_id_ctx so every log line emitted during the request carries the correlation ID.
• apply_cors(app, settings) reads CORSSettings defaults; pass keyword overrides for one-off changes.

• register_exception_handlers(app) wires three handlers, each with its own log level:

  • AppException → {detail, code, details} envelope + INFO log (4xx) or ERROR + traceback + 500.log (5xx).
  • HTTPException → keeps Starlette's default body ({"detail"}) on 4xx with an INFO log; on 5xx swaps in the SDK envelope + traceback + 500.log.
  • Exception (catch-all) → SDK envelope + traceback + 500.log (fixes Starlette's default, which returns only "Internal Server Error" with no log entry).

  Every handler honors RequestIDMiddleware: the log line carries the request_id, and the envelope exposes it under details so the client can correlate. Pass log_traceback=False when an APM (Sentry, OpenTelemetry) is already capturing the stack trace. - make_health_router(db=db, checks={"redis": redis.health_check}, version=...) mounts GET /health/liveness and GET /health/readiness (returns 503 when any check fails) at the root prefix. - make_token_dependency(secret) returns an async dependency that validates X-Token via hmac.compare_digest; pass an empty string to disable in dev. The dependency lives next to the rest of the auth glue in src/api/dependencies/auth.py once it grows beyond the one-liner above.

Localized error messages (i18n)

By default the envelope detail is the exception's literal message (English for the built-ins). To return the message in the client's language without translating at each raise, pass a MessageCatalog to 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(),                   # ← bundled PT-BR + EN-US
        default_locale="pt-BR",
    )
    ...

The handler negotiates the locale from the Accept-Language header (ordered by q), falls back to default_locale when nothing matches, and resolves the exception key — message_key if set, else the code — against the catalog. With no catalog, or when the key is unknown, the literal detail is kept (zero breakage).

# Same NotFoundException, language decided by the client's Accept-Language:
#   Accept-Language: pt-BR  →  {"detail": "Recurso não encontrado", "code": "NOT_FOUND"}
#   Accept-Language: en-US  →  {"detail": "Resource not found",     "code": "NOT_FOUND"}

For domain codes (and messages with parameters), extend the catalog with merge and pass message_params at the 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",                                    # literal fallback
        code="USER_NOT_FOUND",
        message_params={"email": email},
    )

The key defaults to the code

You rarely pass message_key — it falls back to the exception's code. Set message_key only to decouple the translated string from the error code. A template referencing a missing param is returned uninterpolated rather than raising.

JWT bearer / current-user / role dependencies

Four dependency factories live in tempest_fastapi_sdk.api.dependencies.auth — pick the level of abstraction you need.

Factory	What you get
make_token_dependency(secret)	Validate the X-Token shared-secret header (constant time).
make_bearer_token_dependency(tokens, soft=False)	Decode Authorization: Bearer <jwt> and return the claims dict.
make_jwt_user_dependency(tokens, user_loader, soft=False, subject_claim="sub")	Decode the bearer JWT, await user_loader(subject), return the loaded user.
make_role_dependency(tokens, ["admin"], require_all=False, roles_claim="roles") / make_permission_dependency(tokens, ["users:write"], require_all=True, permissions_claim="permissions")	Decode the bearer JWT and gate the route on roles / permissions.

Outside a route: require_x_token

The shared-secret check also exists in imperative form — require_x_token(secret, token) raises UnauthorizedException on a mismatch (and is a no-op with secret="", the dev mode). Use it where there is no Depends to hang off: a queue consumer receiving the token in the payload, a WebSocket handler, a script. On a route prefer make_token_dependency(secret) — Swagger then shows the 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", ""))

Using the bundled flow? Skip load_user

If you wire auth with UserAuthService + make_auth_router, you don't write load_user or instantiate a JWTUtils here — call auth_service.current_user_dependency() (and .current_user_dependency(soft=True)), which reuses the service's internal JWTUtils. See the auth recipe ». The example below is the manual wiring, for when you're not using the 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 returns None instead of raising on missing/invalid tokens — useful for endpoints that work both authenticated and anonymous. subject_claim defaults to "sub" but can be any custom claim ("user_id", "uid", ...). Role dependencies accept either a string or a list of strings on the JWT claim; require_all=True requires every listed role/permission, False (default for roles, override for permissions) requires any.

Rate limit middleware

RateLimitMiddleware is a sliding-window limiter — each unique key (client IP by default) is allowed at most max_requests requests inside every window_seconds window. Exceeded requests get a 429 Too Many Requests with a Retry-After header and the SDK's canonical error envelope as the body (see below). Two axes are pluggable: the store (memory or Redis) and the key (IP, user, tenant, API key) — see below.

# 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"),
    )
    ...

Limit per user / tenant / API key

By default the key is the client IP. To limit per principal (authenticated user, tenant, API key), pass a key_func. The SDK ships ready-made factories:

Factory	Key produced	Use
key_by_ip(trusted_header=...)	ip:<addr>	Per IP (default).
key_by_jwt_subject(jwt)	user:<sub>	Per authenticated user (sub claim).
key_by_jwt_claim(jwt, "tenant_id", scope="tenant")	tenant:<id>	Per arbitrary token claim.
key_by_header("x-api-key", scope="apikey")	apikey:<value>	Per header value.

The middleware runs before dependencies

RateLimitMiddleware runs before FastAPI Depends resolve — so the user authenticated by your auth dependency does not exist yet when the key is computed. That is why the key_by_jwt_* factories decode the bearer from the raw request (via JWTUtils.decode_or_none, no exception raised). Anonymous traffic falls back to the IP, so it stays limited.

# 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()),        # ← limit per user
        exempt_paths=("/health/liveness", "/health/readiness"),
    )
    return app

Distributed state with Redis

The default store (MemoryRateLimitStore) counts in-process — correct for a single worker. For multi-replica deployments, pass store=RedisRateLimitStore(redis): each key becomes a sorted set and a single Lua script prunes expired members, counts, and adds the new hit atomically (no race between count and add). On a Redis error, fail_open=True (default) allows the request rather than locking everyone out.

# 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),                    # ← shared across replicas
        exempt_paths=("/health/liveness", "/health/readiness"),
    )
    return app

The sliding-window semantics are identical across both stores; only where the counters live changes. You can still push rate limiting to the edge (nginx / Cloudflare / AWS WAF) when you prefer.

Why Redis.from_url here, not AsyncRedisManager?

This client feeds a middleware, built in create_app (sync), before any async lifespan runs. Redis.from_url() is lazy — it constructs without opening a connection, so it fits here. AsyncRedisManager needs await connect() and fits where there's an async context: a client via Depends(cache.client_dependency), or the SSEBroker built in the lifespan. Both need the [cache] extra (the redis package).

Tolerating bursts: the token bucket

A sliding window answers one question: did this key exceed N requests in the last W seconds? A client that fires 20 requests at once and then idles for a minute is well-behaved on average — and a strict window rejects it anyway.

A token bucket separates the two: the sustained rate (tokens refilled per second) and the burst it absorbs (bucket capacity). A RateLimitRule with burst is a token bucket; without it, it stays a sliding window.

from tempest_fastapi_sdk import RateLimitRule

# 10 req/s sustained (600 per minute), absorbing a burst of 100.
rule = RateLimitRule(max_requests=600, window_seconds=60.0, burst=100)

Rules reach the middleware through a policy. For a single-tier service, StaticRateLimitPolicy applies one list to everyone:

# 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= and store= are alternatives

max_requests / window_seconds / store describe one sliding window. With policy= you pass a list of limits, and the counters live in quota_store= (MemoryQuotaStore by default, RedisQuotaStore for multi-replica). Passing both raises ValueError at construction — honoring one would silently ignore the other.

Per-plan quotas

One rule list per tier, resolved per request. A plan's rules are checked together: a per-minute limit stacked under a daily ceiling.

# 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="minute"),
                RateLimitRule(1_000, 86_400.0, scope="day"),
            ],
            "pro": [
                RateLimitRule(600, 60.0, burst=100, scope="minute"),
                RateLimitRule(100_000, 86_400.0, scope="day"),
            ],
        },
        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

Three decisions worth understanding:

• An unknown plan falls back to default_plan. A tier the configuration does not know must be limited, not served unlimited — and a request is the wrong place to discover a typo. What is validated at construction: an empty plans, a default_plan outside the mapping, and a plan with no rules. All three would only show up in production as unlimited traffic.
• A rejection spends nothing. A request blocked by the daily ceiling must not burn a token from the per-minute limit — otherwise the minute drains without a single request being served. Both stores decide every rule before writing any of them; RedisQuotaStore does it inside one Lua script, which is the only way to keep the whole list atomic.
• key_by_plan_principal prefixes the plan onto the key. An upgrade starts writing to fresh counters instead of inheriting the exhausted ones. The price is that a caller who chooses their own plan (a client-supplied header) could reset their counters by switching — pair it with a resolver your edge controls.

The 429 body is the SDK's error envelope

The 429 comes out in the same shape register_exception_handlers writes in every handler, so a client parses one envelope for every failure:

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

code is what a client branches on -- never detail, which is prose and changes with the negotiated locale once a MessageCatalog is registered. Swap both for whatever your API uses:

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="Slow down, you are sending too many requests.",
    error_code="TOO_MANY_REQUESTS",
)

details carries what only headers had before: retry_after_seconds (the same number as Retry-After) and the limit of the rule that rejected the request -- in policy mode, the tightest one.

Changed in v0.256.0

Up to v0.255.0 the 429 was text/plain with the raw error_message as the body. Clients reading that body as text have to read JSON and take detail instead; clients branching on status === 429 are unaffected.

The swap closes a contradiction inside the SDK itself: error_responses() always pointed 429 at ErrorResponseSchema, so a client generated from the OpenAPI schema broke deserializing the text.

Documenting the 429 in OpenAPI

error_responses(TooManyRequestsException) describes exactly what the middleware sends -- the default error_code is that exception's own code. Change error_code and you should change the exception documenting the route too, or the schema drifts from the body again.

Why the middleware does not raise the exception

That would be the obvious route -- raise TooManyRequestsException and let the registered handler format it. It does not work: a BaseHTTPMiddleware added through add_middleware sits outside Starlette's ExceptionMiddleware, so an exception raised in dispatch finds no handler and turns into a 500. The middleware therefore builds the response itself, reading the default code off that same exception so the two cannot drift.

RateLimit-* headers

Every response carries RateLimit-Limit and RateLimit-Remaining (turn them off with limit_headers=False), describing the tightest rule — the one a client should pace itself against. RateLimit-Reset is emitted only when the number is known: always in policy mode, and only on 429 in single-window mode. The sliding-window store reports no reset for an accepted request, and a guessed number is worse than an absent header.

Multi-replica

# 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),             # ← shared across replicas
    )
    return app

A window rule becomes a sorted set of timestamps; a bucket rule becomes a hash holding tokens + ts. The bucket key's TTL covers the time to fill the bucket, not the window: a slow bucket with a large burst (10 tokens/minute, burst=1000) takes over an hour to fill, and expiring before that would hand back a full bucket for free. On a Redis error, fail_open=True (default) allows the request.

Request body size limit (BodySizeLimitMiddleware)

A 2 GB upload against an endpoint that expects JSON kills the worker before Pydantic validates anything. BodySizeLimitMiddleware is pure ASGI and stops it at the door:

# 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 for any request
    exclude_paths=("/api/files/upload",),  # the upload route caps itself
)

There are two checks:

1. Header — a Content-Length above the cap answers 413 immediately, without reading a single body byte. This covers the common case where the client knows the size.
2. Streaming — on a chunked upload (no Content-Length), the middleware counts the bytes of the http.request messages and aborts once the cap is crossed.

The response is the SDK's canonical envelope: {"detail": "Request body too large.", "code": "REQUEST_BODY_TOO_LARGE", "details": {"max_bytes": 1048576}} with HTTP 413.

exclude_paths matches by prefix

The match is startswith, so the more specific the prefix the better — ("/api/files",) exempts everything under /api/files. Use it for the route that deliberately accepts a large file and enforces its own limit (via UploadUtils(max_size=...), say). max_bytes=0 disables the check — never ship that to production.

HTTP response cache (ETag / 304)

ResponseCacheMiddleware delivers two performance wins, layered.

ETag + conditional GET (always on). Every cacheable response gets a strong ETag (a hash of the body) and a Cache-Control. When the client sends a matching If-None-Match, the middleware answers 304 Not Modified with no body — the handler still runs, but the bytes never go over the wire.

from fastapi import FastAPI

from tempest_fastapi_sdk import ResponseCacheMiddleware


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

Server-side cache (opt-in via store=). With a store wired, a cacheable GET/HEAD response is stored for ttl_seconds; a later matching request is served without running the handler (X-Cache: HIT), and an If-None-Match hit against the stored ETag still short-circuits to 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),   # shared across replicas
        ttl_seconds=60,
        vary=("Accept-Encoding",),               # varies the key and emits Vary
    )
    return app

Only safe methods (GET/HEAD) and successful responses (200 by default) are cached. Responses that opt out — Cache-Control: no-store/private or a Set-Cookie (personalized) — are never stored.

Cache key

The key is method|path|query plus the headers listed in vary=. Pass cacheable=<predicate> to exclude specific requests, or exempt_paths=(...) to skip exact paths. The store mirrors the idempotency one (memory or Redis with a raw client), so it composes with the service's existing Redis.

A credentialed request does not share a cache entry

Notice what the key does not contain: who asked. If two people's GET /api/me land on the same key, the first stored response is served to the second — and to any anonymous caller hitting the same path.

So a request carrying Authorization or Cookie bypasses the shared store. It still gets an ETag and a 304 (both per-response, and safe); it just never enters a cache someone else can read.

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: the key now includes the credential
    )
    return app

With cache_credentialed=True a digest of the credential headers joins the key, so each caller gets its own entry and caching works for authenticated routes again. Only turn it on after checking that the cached routes do not carry a second identity the credential does not stand for (a tenant header, a session independent of the path) — there, credentials alone do not separate enough.

Cache-Control defaults to private

The emitted default is private, max-age=<max_age>. The middleware cannot know whether a route's body is personalized, and a wrong public there makes a shared proxy serve one user's response to another. Pass cache_control="public, max-age=…" deliberately, on a router that only serves shared content.

Webhook signature verification

WebhookSignatureVerifier validates HMAC-signed inbound webhooks (Stripe / GitHub style) and exposes a FastAPI dependency that reads the raw body, checks the signature with hmac.compare_digest, and yields the body bytes so the route handler can re-parse it without re-reading the 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)
    ...

Supports hex (default) and base64 encodings, any hashlib algorithm guaranteed across platforms, and an optional prefix (e.g. "sha256=") stripped before comparison. Use the imperative verifier.verify(body, signature) from queue handlers when validation happens outside the FastAPI pipeline.

The signature covers the body only, so a captured delivery stays valid forever — anyone who observes one can resend it indefinitely. Pass timestamp_header= to also require a recent unix timestamp and reject anything outside the window:

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,
)

It is opt-in because a provider that sends no such header would have its legitimate traffic rejected. The SDK's own WebhookSender sends X-Webhook-Timestamp.

The timestamp is not covered by the signature

Since only the body is signed, whoever replays a capture can also rewrite the timestamp to a current one. The check bounds accidental and opportunistic replay, not an active attacker who reads and rewrites the request. Closing that depends on the provider signing timestamp.body — the shape Stripe uses; when yours does, verify it with verifier.verify(f"{ts}.".encode() + body, signature).

For providers that sign with an RSA private key (Apple App Store, Google Play, custom enterprise services), swap WebhookSignatureVerifier for RSAWebhookSignatureVerifier — same verify(body, signature) surface, but it validates the signature against a PEM-encoded public key. Uses RSASSA-PKCS1-v1_5 over SHA-256/384/512 (configurable via algorithm=). Requires the cryptography package (installed by the [webpush] extra).

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",
)

# From queue handlers / outside FastAPI:
ok: bool = apple.verify(raw_body_bytes, base64_signature_header_value)

Outbound webhook delivery — WebhookSender

The counterpart: sending signed events to your subscribers. WebhookSender POSTs the JSON event, signs the body with the same WebhookSignatureVerifier (so the receiver validates with that verifier) and retries transient failures (connection error, 5xx, 429) with exponential backoff. Other 4xx are not retried. The httpx client is injected (you own its lifecycle).

Installation

The rest of the HTTP layer ships with tempest-fastapi-sdk. WebhookSender needs the [http] extra — uv add "tempest-fastapi-sdk[http]" (pulls in 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://subscriber.example.com/hooks",
            event="order.paid",
            payload={"id": order["id"], "total": 4200},
        )
        if not result.delivered:
            # result.status_code / result.attempts / result.error
            ...  # enqueue for retry, alert, etc.

    # Same event to many subscribers, concurrently:
    results = await sender.send_many(
        [(url, {"id": order["id"]}) for url in subscribers],
        event="order.paid",
    )


asyncio.run(main())

Each delivery sends X-Webhook-Event, X-Webhook-Id (a unique uuid) and X-Webhook-Timestamp headers, plus the HMAC signature in the signer's header. Returns a WebhookDelivery (delivered, status_code, attempts, error, delivery_id).

Pairs with the outbox

Combine with BaseOutboxModel + OutboxRelay: write the event in the same transaction as the business change and let the relay call WebhookSender — at-least-once delivery with the signature the subscriber verifies.

Pagination Link headers

build_pagination_link_header emits an RFC 8288 Link header with first / prev / next / last rels — pair it with (or use instead of) the BasePaginationSchema body wrapper for REST clients that expect GitHub-style headers. Existing query parameters on the base URL are preserved.

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

Tweak page_param= / size_param= when your service uses non-standard query parameter names (e.g. offset / limit). Pass extra_params={"sort": "name"} to bake the current sort/filter state into every link.

Tool-spec router

make_tool_spec_router(spec) mounts a GET /tool-spec endpoint exposing a machine-readable manifest at the root prefix — meant to sit alongside /health/liveness so external callers can discover capabilities without parsing the full OpenAPI document.

# 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

Pass a dict (served verbatim), a sync callable (called every request) or an async callable (awaited). Override path= to expose the manifest at a different URL or tag= to group it under a different OpenAPI tag.

Programmatic server entry point

run_server is the canonical helper imported from src/server.py. It centralizes the host / port / reload defaults — pulling values from a ServerSettings-flavoured settings object when present — and keeps the entry point a single line.

# 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()

Resolution order for each kwarg is explicit argument → settings.SERVER_* → SDK default ("127.0.0.1" / 8000 / False). Extra uvicorn kwargs (workers=, log_config=, ssl_*=) are forwarded verbatim.

Settings mixins composition

BaseAppSettings is the configured pydantic-settings base (env_file=".env", extra="ignore", case_sensitive=True). The SDK also exposes composable mixins for the most common dependencies; pick the ones the service needs and compose them on your Settings class.

BaseAppSettings must be the last base

Every mixin inherits BaseAppSettings, so they all carry the same model_config — including env_file=".env" — and .env is read regardless of the order in which the mixins appear. But BaseAppSettings must be the last base: since the mixins subclass it, listing it before any mixin violates Python's C3 linearization and breaks the import. As of v0.159.1 the AppSettingsMeta metaclass intercepts this and raises TypeError: Settings: BaseAppSettings must be the LAST base — DatabaseSettings already subclasses it ..., which names the fix; before that the message was pydantic's raw TypeError: Cannot create a consistent method resolution order (MRO). Always put BaseAppSettings at the end of the bases — see the migration guide.

# 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()

Each mixin owns its own env-var prefix — pick only the ones the service needs:

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

Breaking change in 0.8.0: ServerSettings previously exposed bare HOST / PORT / DEBUG / LOG_LEVEL / LOG_JSON fields. They were renamed to SERVER_HOST / SERVER_PORT / SERVER_RELOAD / SERVER_DEBUG, and LOG_LEVEL / LOG_JSON moved to the new LogSettings mixin. Update both your .env file (env var names) and any code reading settings.HOST etc.

Authentication

End-to-end signup + login + protected route using PasswordUtils and JWTUtils. Requires the [auth] extra.

Wire the utility singletons

# 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

Reuse the UserService.create defined in the tutorial — it already hashes the password.

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/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)

Protect a route — JWT dependency

Use make_jwt_user_dependency to wire the bearer scheme + JWT decode + user load in one call. The single seam is the user_loader, an async callable that maps the JWT subject claim to your domain UserModel — called as user_loader(subject) by default, or user_loader(subject, session) once you pass session_dependency= (recommended: the user then comes back attached to the request session and can be mutated without an InvalidRequestError).

The token is looked up in header → cookie → query string, first hit wins: Authorization: Bearer always, plus the cookie_name= cookie and the query_param= query parameter when you enable them (both None by default). subject_claim= changes which claim is read (default "sub") and error_message= the 401 text.

Returning None from user_loader refuses the request (v0.252.0)

The loader is the natural place to refuse a user — deactivated account, a subject that no longer exists, a malformed id. With soft=False (the default), None becomes a 401, the same answer an absent subject already got.

Up to v0.251.0 that None reached the handler and the route answered 200 with a user the loader had rejected — so deactivating an account had no effect at all until the access token expired. With soft=True the None still arrives: that is what the flag is for.

Raising UnauthorizedException or NotFoundException from inside the loader still works, and is what you want when the message matters.

# 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)

Soft auth (optional user)

get_current_user_or_none above already uses soft=True — it returns None instead of raising on a missing or invalid token, so endpoints can work both authenticated and anonymous:

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)

Under the hood soft=True calls tokens.decode_or_none (no exception on expired/invalid tokens) and skips the loader when the subject is missing.

---

File uploads

Avatar endpoint with validation + cleanup. Requires the [upload] extra.

# 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 reads the first bytes of each upload and confirms the file really is one of the allowed types — an HTML+JS payload sent as image/png is rejected even though its extension and Content-Type header look valid. Only enable it when every accepted format is one sniff_mime recognizes (JPEG, PNG, GIF, BMP, WebP, PDF); otherwise a legitimate but unsniffable upload would be refused. For finer control, pass a content_validator predicate to save() (save(file, content_validator=lambda b: sniff_mime(b) in {"image/png"})), and pass filename="..." for a deterministic, addressable name (e.g. f"{user_id}.jpg") instead of the default UUID.

# 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))

Add set_avatar to both the service and the controller (the controller stays a thin pass-through unless orchestration is needed — e.g. firing an "avatar updated" event):

# 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() raises FileTooLargeException (413) or InvalidFileTypeException (415) on rejection — the SDK's exception handler already returns the right status code with a code field on the response.

Serving the file back

Local-disk uploads are best served by an upstream (nginx / Caddy) so FastAPI doesn't stream bytes. For 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",
)

Construct the public URL in the response schema:

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}"

Serving private files through the API (DownloadUtils)

When a file must stay behind auth — invoices, contracts, medical scans — a public /static URL leaks it to anyone who learns the path. DownloadUtils streams the bytes through the endpoint itself, so the same Depends(get_current_user) / permission checks that guard every other route guard the download too. No public link is ever exposed. It needs no extra (uses Starlette's FileResponse / StreamingResponse, which ship with 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
    )

Any relative path that escapes base_dir (../ traversal, absolute paths, symlink escapes) raises NotFoundException (404) instead of leaking the file — the same 404 you get for a genuinely missing file, so callers never distinguish "forbidden" from "absent". file_response guesses the MIME type from the filename (override with media_type=), and as_attachment=False serves inline (e.g. preview a PDF in-browser).

For payloads built on the fly — a generated report, an in-memory zip, decrypted bytes — use stream() instead of touching disk:

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() accepts raw bytes, a sync Iterable[bytes], or an AsyncIterable[bytes], so a large export can be yielded chunk-by-chunk without buffering it all in memory. Both methods set a UTF-8-safe Content-Disposition (non-ASCII filenames survive via the RFC 5987 filename* parameter); build_content_disposition() is exported if you need to set that header on a hand-rolled response.

---

Transactional email

Password reset flow using EmailUtils + a short-lived JWT. Requires the [email] extra.

# 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 / next steps

You now know the whole HTTP surface: app bootstrap, exception handlers with i18n, auth dependencies, rate limiting, webhook verification, pagination headers, tool-spec, the server entry point and settings composition. The last three sections (authentication, uploads, email) were a summary — each has a dedicated recipe that goes deeper:

• Authentication flow » — the full bundled flow (UserAuthService + make_auth_router): signup, activation, login, password reset, refresh tokens and current_user.
• File uploads » — pluggable storage backends (LocalUploadStorage / MinIOUploadStorage), presigned URLs and content validation.
• Transactional email » — EmailUtils with Jinja2 templates, the bundled defaults (activation.html, password_reset.html) and how to shadow them.
• Private downloads » — serving files behind auth with DownloadUtils (file_response / stream), without leaking public links.