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.
• make_rate_limit_dependency — the same limit on one route, with a key that can read the body (key_by_body_field).
• 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 (plus a fourth, for the 422, when envelope_validation_errors=True — see The 422 inside the envelope):

  • 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 — see Liveness and readiness. - 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.

Liveness and readiness

They are two different questions, asked by whoever decides two different things:

• Liveness — "is the process alive?". When it fails, the orchestrator (Kubernetes, ECS) restarts the container. That is why GET /health/liveness touches no dependency at all: a Redis outage is not fixed by restarting the API, and a liveness probe that depended on it would take every replica down at once, in a loop.
• Readiness — "can I take traffic right now?". When it fails, the load balancer takes the replica out of rotation and puts it back once it passes again. This is where the database, cache and broker belong: GET /health/readiness runs every check and answers 503 when any fails.

A single GET /health mixes the two — it either restarts over a dependency or sends traffic to a replica that cannot serve it. When you ask for the health endpoints of a new service, ask for both.

import asyncio

from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient

from tempest_fastapi_sdk import make_health_router


async def redis_check() -> bool:
    """Answer right away: a healthy dependency."""
    return True


async def search_check() -> bool:
    """Hang, like a half-open socket that never answers."""
    await asyncio.sleep(60)
    return True


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


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


asyncio.run(main())

Output (the whole process took 1.7 s, despite the sleep(60)):

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

What happens in the readiness probe:

• Checks run concurrently, each bounded by timeout (default 3.0 seconds; None disables it). One that overruns counts as a failure for itself only — redis above stays True — and the response leaves right after the timeout, not when the orchestrator's probe gives up. Keep timeout below the probe timeout.
• The log never carries the exception message. A check that raises logs Health check 'database' raised ConnectionRefusedError — name and type, never the text, because database and broker driver messages often carry the DSN with its user and password.

Cancellation only happens at an await

timeout cancels the check at its next await point. A check that blocks the event loop with synchronous I/O (a non-async driver, a time.sleep) cannot be interrupted — and it stalls the whole process, not just the readiness probe. Run that kind of call with asyncio.to_thread.

Queue

MessageBroker.health_check runs FastStream's real ping, with timeout=2.0 by default — below the router's 3 s. With RabbitMQ stopped after boot it returns False: make_health_router(checks={"queue": mq.health_check}).

Every 4xx inside the envelope

The 422 was the most visible gap, not the only one. The other four client-error categories do reach this SDK's handler — six cases, each measured in the table below — and answered Starlette's bare body, with no code and in English, even in a service with a configured MessageCatalog:

Path	Before	With envelope_client_errors=True
raise HTTPException(404, "order 42 does not exist")	{"detail": "order 42 does not exist"}	same message plus code: NOT_FOUND
raise HTTPException(403)	{"detail": "Forbidden"}	Acesso negado + code: FORBIDDEN
raise HTTPException(409)	{"detail": "Conflict"}	Conflito de recurso + code: CONFLICT
Depends(HTTPBearer()), no token	{"detail": "Not authenticated"}	same message plus code: UNAUTHORIZED
unknown route	{"detail": "Not Found"}	Recurso não encontrado + code: NOT_FOUND
wrong method	{"detail": "Method Not Allowed"}	same message + code: HTTP_405

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import default_message_catalog, register_exception_handlers


def create_app() -> FastAPI:
    """Build the application.

    Returns:
        FastAPI: The configured application.
    """
    app: FastAPI = FastAPI()
    register_exception_handlers(
        app,
        catalog=default_message_catalog(),
        default_locale="pt-BR",
        envelope_client_errors=True,       # ← every 4xx, the 422 included
    )
    return app

A message you wrote is not translated

This is the line between localizing and losing information. Starlette fills detail with the status phrase when the caller passes none, so the two cases are distinguishable:

• HTTPException(404, "order 42 does not exist") → your message, preserved
• HTTPException(404) → detail == "Not Found", the phrase → translated

Replacing the first with the catalog's generic entry would destroy the more informative of the two. FastAPI's own "Not authenticated" for a missing bearer token falls in the first group too — measured — and is preserved.

Why two flags rather than one rename

envelope_validation_errors shipped in v0.284.0 covering only the 422. envelope_client_errors is the superset: it turns on every 4xx, the 422 included. A consumer who had already opted into the first sees no behaviour change on upgrade — which is exactly what a silent rename would cause.

Want everything? envelope_client_errors=True alone. Want only the 422? The other one alone.

HTTP_<status> for a status the SDK does not model

The status-to-code mapping is declared, not derived: measured, thirteen AppException subclasses carry a 401 code and four a 404, so there is no status to read a single code off.

A status outside the mapping answers HTTP_<status> and keeps the framework's phrase. 400 and 405 are in that group today — no AppException subclass carries a code for either. A client still parses one shape; what is missing is the translation, not the code.

The 422 inside the envelope

register_exception_handlers wires three handlers, and the 422 goes through none of them: FastAPI intercepts RequestValidationError in its own default handler. So the error a client sees most often was the only one outside the SDK's error model — no code, and never through the MessageCatalog the service configured.

# src/api/app.py

from fastapi import FastAPI

from tempest_fastapi_sdk import default_message_catalog, register_exception_handlers


def create_app() -> FastAPI:
    """Build the application.

    Returns:
        FastAPI: The configured application.
    """
    app: FastAPI = FastAPI()
    register_exception_handlers(
        app,
        catalog=default_message_catalog(),
        default_locale="pt-BR",
        envelope_validation_errors=True,        # ← put the 422 in the envelope
    )
    return app

With the flag on, POST /login with a short password answers:

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

The default 422 sends the submitted value back — the password included

The body FastAPI generates includes input for every field that failed. Measured, with password failing min_length:

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

And SecretStr does not protect it: validation runs before the secret wrapper exists, so a failing SecretStr field returns the raw value too. Measured in both cases.

That is the strongest reason to turn the flag on. details carries loc and type and never input — this is not a warning for you to act on, it is a line the handler does not write.

The table is keyed by error type, not by field

Messages come from the catalog under VALIDATION.<pydantic type> — string_too_short, missing, int_parsing and the other 101. That is pydantic's vocabulary, identical in every service, so a new field on a schema is born translated: there is no per-field table to keep.

There are 104 types, exactly the set of pydantic_core.ErrorType, and tests/test_pydantic_error_types_guard.py fails when the installed pydantic and the table disagree in either direction.

The catalog carries Portuguese only, on purpose: pydantic's own msg already is the English message, so the handler falls back to it and the SDK keeps no copy that could drift from upstream. A type the catalog does not know falls back the same way — it degrades to English, never to blank.

The SDK's BR validators localize the whole phrase

Pydantic files everything a field validator raises under the single type value_error, with the exception in ctx["error"]. Since the catalog localizes by type, CPFOrCNPJField, PhoneBRField, UFField, postal codes and PIX keys arrived half-translated:

Valor inválido: invalid CPF/CNPJ

As of v0.290.0 those validators raise ValidationValueError, which carries a code (INVALID_CPF_CNPJ, INVALID_CEP, INVALID_UF, …) resolved as VALIDATION.<code> in both languages — CPF ou CNPJ inválido in pt-BR, Invalid CPF or CNPJ in en-US. English is carried here, unlike the pydantic table: upstream's msg for this case is Value error, invalid CPF/CNPJ, prefixed and lowercase.

Your own validators are unaffected: a plain ValueError still falls back to VALIDATION.value_error, and ValidationValueError subclasses ValueError, so every except ValueError keeps catching it.

Turning the flag on changes the body, so it is opt-in

A frontend doing body.detail.map(...) and a client generated from the OpenAPI schema both model the 422 as an array. That is why the default is False, and turning it on is the service's decision.

The schema follows: with the flag on, the HTTPValidationError component is rewritten to the envelope (detail as a string, plus code and details) and the item loses input. Every route references that component, so the docs and the runtime agree in both positions of the flag — without it the schema would lie in exactly the response a generated client uses most.

Telling someone, and choosing where the record lands

log_level decides the severity of a 5xx. Two things it does not decide: who finds out, and which logging configuration the record lands in.

# src/api/app.py

import logging

from fastapi import FastAPI, Request

from tempest_fastapi_sdk import EmailUtils, LogUtils, register_exception_handlers

from src.core.settings import settings

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


async def alert_ops(request: Request, exc: Exception) -> None:
    """Alert operations about a 5xx, off the response path.

    Args:
        request (Request): The request that failed.
        exc (Exception): The exception being reported.
    """
    mailer: EmailUtils = EmailUtils(**settings.email_kwargs())
    await mailer.send(
        to=[settings.OPS_EMAIL],
        subject=f"5xx on {request.url.path}",
        body=f"{type(exc).__name__}: {exc}",
    )


def create_app() -> FastAPI:
    """Build the application.

    Returns:
        FastAPI: The configured application.
    """
    app: FastAPI = FastAPI()
    register_exception_handlers(
        app,
        logger=log.logger,                # ← where the record lands
        on_server_error=alert_ops,        # ← who finds out
        log_level=logging.ERROR,
    )
    return app

on_server_error runs as a background task

Called with (request, exc) after the response is built, so it never delays or alters what the client receives. It fires on all three paths that produce a 5xx — the catch-all, HTTPException(500) and a 5xx AppException — and does not fire on 4xx: an alert that pages on a 404 is an alert nobody reads.

An exception inside the callback is logged and swallowed

On purpose. Measured without that protection: the 500 response is still correct, but the notifier's exception propagates up the ASGI stack and becomes what the server reports — the client's 500 turns into a worse 500 with no trace of the real cause. With it, what propagates is the original exception.

logger= matters when logging is scoped to a logger

The handlers log to their own logger (tempest_fastapi_sdk.api.handlers). With LogUtils(..., scope="root") — the default — that already reaches the service's 500.log and error.log; measured, one line in each.

With scope="logger", which does not configure the root, the SDK's logger sits outside the configured tree and the record reaches none of the files — measured, 0 and 0. Passing logger=log.logger closes that: 1 and 1.

The type is logging.Logger rather than the RetryLogger protocol because the handlers pass extra= and exc_info= and the protocol has neither. Measured: LogUtils.error(extra=...) stores a field named extra, and logging.Logger.error(request_id=...) raises TypeError — no single type spans the two. LogUtils exposes .logger, which is the object that fits.

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.

JWTUtils built in create_app

With get_settings() and create_app(settings), the JWTUtils only exists inside create_app — and the module-level dependency (CurrentUser = Annotated[...]) is declared before it. Pass a function taking the Request instead of the helper: it is called on every request. It works for make_jwt_user_dependency, make_bearer_token_dependency, make_role_dependency and make_permission_dependency.

from typing import Annotated

from fastapi import Depends, FastAPI, Request

from tempest_fastapi_sdk import JWTUtils, make_jwt_user_dependency


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


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


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


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

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

    return app

Measured in tests/api/test_lazy_dependencies.py: two apps in one process, with different secrets and the same CurrentUser, each refuses (401) the other's token.

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 when Starlette builds the middleware stack — at lifespan startup, not on the add_middleware line, which returns without error. 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.

Clock: Redis's, not the replica's

Bucket refill and window pruning depend on "how much time passed". With several replicas, the question is: by whose clock?

Both Redis stores (RedisQuotaStore and RedisRateLimitStore) answer "by Redis's": the Lua script reads TIME from the server itself, and no replica sends a timestamp. Replicas with out-of-sync wall clocks — or a VM that just took an NTP step — still see one bucket and one window.

What this prevents, measured

Up to v0.301.0 each replica sent its own time.time(). Against real Redis 4 and 7, with two clients — one exhausting the limit, the other probing:

• a client one hour ahead got through a bucket (1 token/hour, burst=3) and a window (3/hour) the other one had just exhausted;
• a client one hour behind wrote timestamps that let the on-time client through the same way.

With the server's TIME, all eight cases are denied. They live in tests/api/test_quota_redis_live.py (make test-docker).

Two practical consequences:

• The clock that matters is the Redis host's. A step there still moves every key at once — but it is one clock, instead of one per replica.
• Redis 4 works, because the script switches to effects replication (redis.replicate_commands()) before writing. Without it Redis 4.0.14 refuses the write after TIME, and under fail_open=True a one-request limit let 10 of 10 through.

redis-py 8 and Redis before 6

redis-py 8.1.0 — the [cache] extra's floor — opens the connection with HELLO 3, which Redis 4 and 5 reject as an unknown command. Against those servers, build the client with protocol=2: Redis.from_url(settings.REDIS_URL, protocol=2).

Testing without depending on the machine's speed

MemoryQuotaStore accepts clock=: any callable returning seconds. With a clock that only moves when the test moves it, "the burst runs out" and "the token comes back after an hour" become exact assertions — not a race against the refill rate:

import asyncio

from tempest_fastapi_sdk import MemoryQuotaStore, RateLimitRule


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

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


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

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


asyncio.run(main())

Output:

[True, True, False]
True

Why not sleep?

A 100-per-second rule refills one token every 10 ms. Four sequential calls on a loaded machine take longer than that, the token comes back, and the "burst runs out" test fails: that is what happened to the SDK's own suite (#339). A negligible rate over the test's span (1 per hour) or an injected clock takes the machine's speed out of the assertion.

Per-route rate limit (make_rate_limit_dependency)

RateLimitMiddleware limits the whole app, and its scope only shrinks by subtraction (exempt_paths). Two questions fall outside it:

• "Only POST /api/invites." An expensive route deserves a ceiling the rest of the app does not have.
• "Per invited e-mail, not just per IP." The key lives in the body, and the middleware's key_func only receives the Request, synchronously, before the body is read.

make_rate_limit_dependency answers both: it is a dependency you hang on one route, it counts in the same RateLimitStore as the middleware (memory or Redis) and it answers the same 429.

Limit one endpoint by IP and by a body field

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

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


class InviteCreateSchema(BaseModel):
    email: str


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

app = FastAPI()
register_exception_handlers(app)


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


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


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

Output:

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

Four different IPs, one e-mail: the fourth request falls on the e-mail key, and a GET on the same URL stays free. Piece by piece:

• key=[...] — each function returns one key (or several, or none), and every key counts in the same store. The first one over budget refuses the request; the ones after it are not counted, so a refused request does not spend their budget. Sync functions (key_by_ip, key_by_header, key_by_jwt_*) and async ones work alike.
• key_by_body_field("email") — reads the field off the JSON body. FastAPI reads the body before resolving dependencies and Starlette keeps the bytes on the Request, so the endpoint still receives its validated payload — the output above is that proof. The value is normalized (strip
• lowercase) and stored as SHA-256, so Redis does not pile up e-mails in clear text (normalize=False, hash_value=False turn that off). A body that is not JSON, or lacks the field, yields no key: the endpoint's own 422 answers that request.
• trusted_ip_header — without key, the default key is the IP resolved from that header (make_rate_limit_dependency(store, ..., trusted_ip_header="x-real-ip")). With key, pass the header to the key_by_ip inside the list; passing both raises ValueError at construction instead of silently ignoring the header.
• The bucket belongs to the route. Each key is prefixed by default with the method and the path template ("POST /api/invites"), so the same dependency on two routes gives each its own budget, /users/1 and /users/2 share the one of /users/{user_id}, and the counter never collides with the middleware's in the same Redis. Pass scope="invites" to make several routes share one budget.

The same 429 as the middleware

The body (detail, code, details.retry_after_seconds, details.limit) and the headers (Retry-After, RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset) come from the same helpers RateLimitMiddleware uses — tests/api/test_rate_limit_dependency.py compares the two. The difference is the path: the dependency raises TooManyRequestsException, so the app needs register_exception_handlers, and with a MessageCatalog registered there detail is translated by code ("Requisições em excesso" under pt-BR), which the middleware — answering outside the handlers — does not do.

A store built in create_app

Pass a function taking the Request instead of the store — make_rate_limit_dependency(store_of, ...), with store_of(request) returning request.app.state.rate_limit_store — and the dependency declared at import time counts in the store of the app serving the request. Measured with two apps and one dependency: each counts in its own.

Multi-replica

Swap MemoryRateLimitStore() for RedisRateLimitStore(redis) and every replica shares the counters, exactly as with the middleware.

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 stream is never drained (and exempt_paths is equality)

Caching means reading the body to the end to hash and store it. In a text/event-stream the body only ends when the client disconnects, so through 0.285.0 an SSE route behind this middleware never answered at all: the request hung until nginx's proxy_read_timeout and became a 504.

Since 0.286.0 the middleware passes a stream straight through. There is nothing to configure: a response with Content-Type: text/event-stream, or with no Content-Length, is neither cached nor given an ETag.

The same was true of IdempotencyMiddleware

It drains the body so it can replay the response on a retry, and it had no exemption list at all. A POST that answers with a stream — such as make_genai_router's /generate/stream — hung the same way whenever the client sent an Idempotency-Key. It passes through now too: a stream is not replayable from any store.

exempt_paths changed meaning in two middlewares. It was equality in ResponseCacheMiddleware, RateLimitMiddleware and GracefulShutdownMiddleware, and prefix in AccessLogMiddleware and HoneypotBanMiddleware — one argument name, two behaviours. A service exempting its SSE route from both passed the same tuple to each and got an exemption in one and a no-op in the other.

Now exempt_paths is equality in all five, and prefix matching has a name of its own:

from fastapi import FastAPI

from tempest_fastapi_sdk import AccessLogMiddleware, ResponseCacheMiddleware


def create_app() -> FastAPI:
    """Build an app whose SSE subtree is exempt from both middlewares."""
    app: FastAPI = FastAPI()
    app.add_middleware(
        ResponseCacheMiddleware,
        ttl_seconds=30,
        exempt_paths=("/health/liveness",),      # this path, and only it
        exempt_prefixes=("/api/sse",),           # and everything under /api/sse
    )
    app.add_middleware(
        AccessLogMiddleware,
        exempt_prefixes=("/api/sse",),           # was: exempt_paths
    )
    return app

Migration: a prefix exempt_paths becomes exempt_prefixes

Only AccessLogMiddleware and HoneypotBanMiddleware change behaviour. If you passed exempt_paths=("/api/sse",) to either of them expecting it to cover /api/sse/stream, rename it to exempt_prefixes — otherwise the exemption stops applying to child routes.

Widening the other way — making everything a prefix — was rejected on purpose: in RateLimitMiddleware an exemption is a hole in the rate limit, and exempt_paths=("/health",) would start exempting /health-admin too, without anyone asking.

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 (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 does not fit this verifier

Stripe-Signature is not a hex HMAC of the body: it is t=<timestamp>,v1=<hex>, and what Stripe signs is "<t>." + body. A WebhookSignatureVerifier(header_name="Stripe-Signature", encoding="hex") rejects every legitimate delivery — measured with a header produced by sign_payload, verify returns False. For Stripe use make_stripe_webhook_dependency from the Stripe integration, which parses the header and checks the time window.

# 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 (and that the Stripe integration already verifies); when your provider 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

WebhookSender uses httpx, which already ships with the base tempest-fastapi-sdk package — no extra is needed. The [http] extra still exists and is optional: it declares httpx explicitly.

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, LOG_DIR, LOG_MAX_BYTES, LOG_BACKUP_COUNT, LOG_REDACT
DatabaseSettings	DATABASE_URL, DATABASE_ECHO, DATABASE_POOL_SIZE, DATABASE_MAX_OVERFLOW, DATABASE_POOL_RECYCLE, DATABASE_SQLITE_WAL, DATABASE_SQLITE_BUSY_TIMEOUT, DATABASE_SQLITE_FOREIGN_KEYS
RedisSettings	REDIS_URL, REDIS_DECODE_RESPONSES
RabbitMQSettings	RABBITMQ_URL, RABBITMQ_PREFETCH_COUNT
TaskIQSettings	TASKIQ_BROKER_URL, TASKIQ_RESULT_BACKEND_URL, TASKIQ_STORE_RESULTS, TASKIQ_RESULT_TTL_SECONDS
JWTSettings	JWT_SECRET, JWT_ALGORITHM, JWT_ACCESS_TTL_SECONDS, JWT_REFRESH_TTL_SECONDS, JWT_ISSUER
CORSSettings	CORS_ORIGINS, CORS_ORIGIN_REGEX, CORS_ALLOW_CREDENTIALS, CORS_ALLOW_METHODS, CORS_ALLOW_HEADERS, CORS_EXPOSE_HEADERS, CORS_MAX_AGE
EmailSettings	SMTP_HOST, SMTP_PORT, SMTP_USERNAME, SMTP_PASSWORD, SMTP_FROM_ADDR, SMTP_USE_TLS, SMTP_USE_SSL, SMTP_TIMEOUT_SECONDS
UploadSettings	UPLOAD_DIR, UPLOAD_MAX_SIZE_BYTES, UPLOAD_ALLOWED_EXTENSIONS, UPLOAD_ALLOWED_MIMETYPES
TokenSettings	TOKEN_SECRET
WebPushSettings	VAPID_PUBLIC_KEY, VAPID_PRIVATE_KEY, VAPID_SUBJECT, WEBPUSH_DEFAULT_TTL_SECONDS

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/dependencies/services.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

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


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

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

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


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


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

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 asks for inline (e.g. preview a PDF in-browser) — granted only to a type in INLINE_SAFE_MEDIA_TYPES, so HTML and SVG still download. Every download response carries X-Content-Type-Options: nosniff, a sandbox CSP and Cross-Origin-Resource-Policy: same-site.

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.