Metadata-Version: 2.5
Name: fastapi-alcazaba
Version: 0.4.0
Summary: Secure, maintained tools for FastAPI: response cache, rate limits, sessions, CSRF, webhooks, a task queue and more
Project-URL: Homepage, https://github.com/nonodev96/fastapi-alcazaba
Project-URL: Documentation, https://nonodev96.github.io/fastapi-alcazaba/
Project-URL: Repository, https://github.com/nonodev96/fastapi-alcazaba
Project-URL: Issues, https://github.com/nonodev96/fastapi-alcazaba/issues
Project-URL: Changelog, https://github.com/nonodev96/fastapi-alcazaba/blob/main/CHANGELOG.md
Author-email: Antonio Mudarra Machuca <nonodev96@gmail.com>
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Keywords: authentication,authorization,cache,caching,casbin,circuit-breaker,cron,fastapi,jwt,oauth2,pagination,rate-limiting,rbac,redis,saml,scheduler,security,toolkit,valkey
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: fastapi>=0.115
Requires-Dist: pydantic>=2.7
Requires-Dist: typing-extensions>=4.10
Provides-Extra: all
Requires-Dist: aiobotocore>=2.13.1; extra == 'all'
Requires-Dist: aiomcache>=0.8.2; extra == 'all'
Requires-Dist: backports-zstd>=1.0; (python_version < '3.14') and extra == 'all'
Requires-Dist: cryptography>=44; extra == 'all'
Requires-Dist: httpx2>=2.15.1; extra == 'all'
Requires-Dist: opentelemetry-api>=1.27; extra == 'all'
Requires-Dist: pycasbin>=2.0; extra == 'all'
Requires-Dist: pyjwt[crypto]>=2.10; extra == 'all'
Requires-Dist: redis>=5.0.8; extra == 'all'
Requires-Dist: signxml>=5.2; extra == 'all'
Requires-Dist: valkey-glide>=2.5; extra == 'all'
Requires-Dist: valkey>=6.0; extra == 'all'
Provides-Extra: auth-strategy-http-bearer
Provides-Extra: auth-strategy-jwt
Requires-Dist: pyjwt[crypto]>=2.10; extra == 'auth-strategy-jwt'
Provides-Extra: auth-strategy-oauth2
Requires-Dist: httpx2>=2.15.1; extra == 'auth-strategy-oauth2'
Provides-Extra: auth-strategy-saml
Requires-Dist: signxml>=5.2; extra == 'auth-strategy-saml'
Provides-Extra: casbin
Requires-Dist: pycasbin>=2.0; extra == 'casbin'
Provides-Extra: crypto
Requires-Dist: cryptography>=44; extra == 'crypto'
Provides-Extra: dynamodb
Requires-Dist: aiobotocore>=2.13.1; extra == 'dynamodb'
Provides-Extra: glide
Requires-Dist: valkey-glide>=2.5; extra == 'glide'
Provides-Extra: memcache
Requires-Dist: aiomcache>=0.8.2; extra == 'memcache'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.27; extra == 'otel'
Provides-Extra: redis
Requires-Dist: redis>=5.0.8; extra == 'redis'
Provides-Extra: valkey
Requires-Dist: valkey>=6.0; extra == 'valkey'
Provides-Extra: zstd
Requires-Dist: backports-zstd>=1.0; (python_version < '3.14') and extra == 'zstd'
Description-Content-Type: text/markdown

<!-- --8<-- [start:header] -->
<p align="center">
  <img src="https://raw.githubusercontent.com/nonodev96/fastapi-alcazaba/main/docs/assets/logo.png" alt="Alcazaba's logo: a fortress wall whose horseshoe-arch gate opens on a bolt" width="150">
</p>

<h1 align="center">Alcazaba</h1>

<p align="center">
  <strong>Secure, maintained tools for FastAPI</strong><br>
  Response cache, rate limits, sessions, CSRF, webhooks, a task queue and more, in one package.
</p>

<!-- --8<-- [end:header] -->

<p align="center">
  <a href="https://nonodev96.github.io/fastapi-alcazaba/"><strong>Documentation</strong></a> |
  <a href="https://nonodev96.github.io/fastapi-alcazaba/latest/examples/">Examples</a> |
  <a href="https://pypi.org/project/fastapi-alcazaba/">PyPI</a> |
  <a href="https://github.com/nonodev96/fastapi-alcazaba/blob/main/CHANGELOG.md">Changelog</a> |
  <a href="https://github.com/nonodev96/fastapi-alcazaba/blob/main/SECURITY.md">Security policy</a>
</p>

<!-- --8<-- [start:badges] -->
<p align="center">
  <a href="https://github.com/nonodev96/fastapi-alcazaba/actions/workflows/ci.yml"><img src="https://github.com/nonodev96/fastapi-alcazaba/actions/workflows/ci.yml/badge.svg?branch=main" alt="CI"></a>
  <a href="https://pypi.org/project/fastapi-alcazaba/"><img src="https://img.shields.io/pypi/v/fastapi-alcazaba?label=PyPI&color=c2552f" alt="PyPI version"></a>
  <a href="https://pypi.org/project/fastapi-alcazaba/"><img src="https://img.shields.io/pypi/pyversions/fastapi-alcazaba?label=Python&color=1b2e42" alt="Python versions"></a>
  <a href="https://scorecard.dev/viewer/?uri=github.com/nonodev96/fastapi-alcazaba"><img src="https://api.scorecard.dev/projects/github.com/nonodev96/fastapi-alcazaba/badge" alt="OpenSSF Scorecard"></a>
  <a href="https://github.com/nonodev96/fastapi-alcazaba/blob/main/LICENSE"><img src="https://img.shields.io/badge/license-Apache%202.0-0e9e8e" alt="License: Apache 2.0"></a>
</p>

<!-- --8<-- [end:badges] -->

---

Alcazaba (`fastapi-alcazaba`) gathers the pieces most FastAPI apps end up writing themselves,
built with security in mind and tested against real servers (the
[examples](https://nonodev96.github.io/fastapi-alcazaba/latest/examples/) show them at work):

| Tool | In `fastapi_alcazaba.` | What it does |
| --- | --- | --- |
| [Response cache](#quick-start) | `response_cache` | `@cache` for endpoints and functions, in memory, Redis, Valkey, memcached or DynamoDB |
| [Rate limiter](#rate-limiting) | `rate_limit` | `FastAPIRateLimiter`: a dependency with a sliding window shared by every process |
| [Locks](#locks) | `background` | `async with cache.lock("job")`: one process at a time, renewed while held |
| [Idempotency keys](#idempotency-keys) | `idempotency` | a retried POST gets the first response instead of running twice |
| [Authentication strategies](#authentication-strategies) | `auth` | `FastAPIAuth`: who makes a request, by strategies: JWTs, bearer tokens, sessions, OAuth2 and SAML logins, your own |
| [Revoked tokens and one-time codes](#revoked-tokens-and-one-time-codes) | `auth` | a JWT blocklist and e-mail/SMS codes, nothing stored in clear |
| [API keys](#api-keys) | `auth` | keys stored only as digests, with scopes and expiry |
| [Authorization with Casbin](#authorization-with-casbin) | `authz` | `FastAPICasbin`: a Casbin policy (ACL, RBAC, ABAC) decides who may do what, per route or for every request |
| [Signed webhooks](#signed-webhooks) | `webhooks` | Standard Webhooks signing and verification, replays refused |
| [Outbound calls](#outbound-calls) | `outbound` | an httpx transport with a deadline per call, a circuit breaker per origin and SSRF protection |
| [Security headers and body size](#security-headers-and-body-size) | `security` | the OWASP headers, and a 413 for endless bodies |
| [CSRF protection](#csrf-protection) | `security` | `CSRFMiddleware`: a 403 for what another site makes the browser send with its cookies |
| [Server-side sessions](#server-side-sessions) | `security` | `request.session` kept in the backend, only a random id in the cookie: revocable |
| [Checking uploads](#checking-uploads) | `uploads` | `FastAPIUploads`: the real type from the bytes, size, image dimensions, safe names |
| [Problem details](#problem-details) | `observability` | errors as RFC 9457 `application/problem+json`, never leaking internals |
| [Request IDs](#request-ids) | `observability` | an ID per request in logs, responses and errors |
| [Audit trail](#audit-trail) | `audit` | `FastAPIAudit`: who did what to which resource, and whether it worked; no secrets, signed and chained |
| [Request timeouts](#request-timeouts) | `observability` | a deadline per request: a 504 instead of a request hung forever |
| [Health checks](#health-checks) | `observability` | `/health/live` and `/health/ready` for load balancers and orchestrators |
| [Scheduled jobs](#scheduled-jobs) | `background` | `FastAPIScheduler`: every N seconds or by cron, each tick once across processes |
| [Task queue](#task-queue) | `background` | `FastAPITaskQueue`: background tasks on Redis streams, retried, at least once, never pickle |
| [Circuit breakers](#circuit-breakers) | `operations` | `FastAPIBreaker`: stop calling a dependency that keeps failing, a 503 meanwhile |
| [Concurrency limits](#concurrency-limits) | `rate_limit` | `FastAPIConcurrencyLimiter`: at most N requests at a time, across processes |
| [Leader election](#leader-election) | `background` | `FastAPILeader`: work that must run in exactly one process at a time |
| [Maintenance mode](#maintenance-mode) | `operations` | `FastAPIMaintenance`: a 503 with `Retry-After` for every process at once |
| [Feature flags](#feature-flags) | `operations` | `FastAPIFeatureFlags`: on, off, by percentage or by subject, shared by every process |
| [Pagination](#pagination) | `pagination` | Twelve strategies: page numbers, offsets, ranges, cursors, keysets, ids, timestamps, sort values, tokens, scrolls and streams, each page cached under its own key |
| [Invalidation by tag](#invalidating-by-tag) | `response_cache` | drop every entry of a product, a user… without knowing their keys |
| [Two-level cache](#two-level-cache) | `backends` | a copy in each process in front of Redis, kept in sync by pub/sub |
| [Metrics and traces](#metrics-and-traces) | `telemetry` | OpenTelemetry for all of the above |

<!-- --8<-- [start:safe-by-default] -->
The response cache is safe by default:

- **Users never get each other's responses.** Requests carrying `Authorization` or `Cookie` get
  their own cache entries and the response is sent as `Cache-Control: private`, also when the
  endpoint authenticates through a router dependency.
- **No Redis script injection.** `FastAPICache.clear(namespace=...)` deletes with `SCAN` +
  `UNLINK`: a namespace from user input never reaches a Lua script, and the server isn't blocked
  by `KEYS`.
- **Signed or encrypted entries.** With a `signing_key`, every entry carries an HMAC bound to its
  key: whoever can write to Redis can't plant data, move one user's entry under another user's
  key, or get a pickle executed. With an `encryption_key` they're also encrypted (AES-256-GCM): a
  dump or a backup of Redis doesn't reveal the cached responses.
- **Bounded memory.** `InMemoryBackend` evicts the least recently used entries instead of
  growing forever with every new query string.
- **Clients can't bypass the cache** with `Cache-Control: no-cache` unless you allow it.
- **Only successful responses are cached**, never a 404 or a 500 replayed as a 200.
- **Cache stampede protection**: concurrent misses for one key compute once, in each process
  and, with `distributed_lock`, across processes.
- **A broken backend doesn't take the app down**: every backend call has a timeout, and a circuit
  breaker stops calling it while it fails.
- **Keys that never repeat are reported**: an argument whose `repr()` changes on every call (a
  database session, a service) makes every request a miss; Alcazaba warns about it.

And more: early refresh of popular entries, Valkey through valkey-py or Valkey GLIDE, cache
hits that return the annotated type (Pydantic models, not dicts), deterministic `ETag`s, the
[`Cache-Status`](https://www.rfc-editor.org/rfc/rfc9211) header (`Alcazaba; hit; ttl=42`,
`Alcazaba; fwd=miss; stored`), the response model in OpenAPI, `exclude_params` and
`stale-while-revalidate`.

<!-- --8<-- [end:safe-by-default] -->

## Install
<!-- --8<-- [start:install] -->

```shell
pip install fastapi-alcazaba            # in-memory backend only
pip install "fastapi-alcazaba[redis]"   # or [valkey], [glide], [memcache], [dynamodb], [all]
pip install "fastapi-alcazaba[otel]"    # OpenTelemetry metrics
pip install "fastapi-alcazaba[crypto]"  # encrypted entries
pip install "fastapi-alcazaba[zstd]"    # zstd compression before Python 3.14
pip install "fastapi-alcazaba[casbin]"  # authorization with Casbin
pip install "fastapi-alcazaba[auth-strategy-jwt]"     # JWTStrategy
pip install "fastapi-alcazaba[auth-strategy-oauth2]"  # OAuth2Strategy
pip install "fastapi-alcazaba[auth-strategy-saml]"    # SAMLStrategy
```

Python 3.11+, FastAPI 0.115+ and Pydantic v2.

<!-- --8<-- [end:install] -->

## Quick start
<!-- --8<-- [start:quick-start] -->

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, Header
from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from redis import asyncio as aioredis

from fastapi_alcazaba import FastAPICache
from fastapi_alcazaba.backends.redis import RedisBackend
from fastapi_alcazaba.response_cache import cache


class Settings(BaseSettings):
    """From environment variables (REDIS_URL...) or a .env file."""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    redis_url: str
    cache_signing_key: SecretStr  # secrets.token_urlsafe(32)


settings = Settings()


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    redis = aioredis.from_url(settings.redis_url)
    FastAPICache.init(RedisBackend(redis), prefix="myapp", signing_key=settings.cache_signing_key)
    yield
    await redis.aclose()


app = FastAPI(lifespan=lifespan)


class Product(BaseModel):
    name: str
    price: float


def current_user(authorization: str = Header()) -> str:
    return authorization  # validate the token here


@app.get("/products/{product_id}")
@cache(expire=60, vary_headers=())  # same response for everybody: one shared entry
async def product(product_id: int) -> Product:
    return Product(name="Olive oil", price=9.5)


@app.get("/me")
@cache(expire=30)  # one entry per Authorization/Cookie value, sent as private
async def me(user: str = Depends(current_user)) -> dict[str, str]:
    return {"user": user}
```

`@cache` goes between the route decorator and the function. It also caches plain functions,
sync or async (sync ones run in the thread pool, like FastAPI does).

The configuration comes from
[pydantic-settings](https://docs.pydantic.dev/latest/concepts/pydantic_settings/) (`pip install
pydantic-settings`). Every parameter that takes a key or a secret also takes pydantic's
`SecretStr` (or `SecretBytes`) as it is, so the value stays out of the settings' repr and of
tracebacks until Alcazaba uses it. The other snippets read theirs from the same `settings`.

<!-- --8<-- [end:quick-start] -->

## Configuration
<!-- --8<-- [start:configuration] -->

### `FastAPICache.init()`

Call it once, at startup (in the `lifespan`).

| Parameter | Default | Description |
| --- | --- | --- |
| `backend` | | `InMemoryBackend`, `RedisBackend`, `ValkeyBackend`, `GlideBackend`, `MemcachedBackend` or `DynamoBackend` |
| `prefix` | `""` | prefix of every key |
| `expire` | `None` | default time to live, in seconds |
| `coder` | `JsonCoder` | how values are stored |
| `key_builder` | `default_key_builder` | builds keys from the function and its arguments |
| `cache_status` | `"Alcazaba"` | this cache's name in the `Cache-Status` header (RFC 9211); `None` to leave it out |
| `enable` | `True` | `False` turns the cache off |
| `signing_key` | `None` | secret of 32+ bytes to sign entries; a list rotates keys (the first signs) |
| `encryption_key` | `None` | secret of 32+ bytes to sign *and encrypt* entries (`crypto` extra); instead of `signing_key` |
| `compression` | `None` | `"zstd"` or `"zlib"`: store large entries compressed |
| `compression_min_size` | `1024` | bytes; smaller entries aren't compressed |
| `compression_level` | codec's default | |
| `vary_headers` | `("authorization", "cookie")` | request headers that give each value its own entry |
| `respect_cache_control` | `False` | let clients skip the cache with `no-cache`/`no-store` |
| `single_flight` | `True` | concurrent misses for one key compute once, per process |
| `backend_timeout` | `1.0` | seconds a backend call may take before the request goes on without cache |
| `circuit_breaker_failures` | `5` | consecutive failures that stop the backend calls; `None` disables the breaker |
| `circuit_breaker_reset` | `30.0` | seconds before trying the backend again |
| `max_value_size` | `None` | bytes; larger values aren't stored |
| `distributed_lock` | `False` | one process computes a missing value while the others wait for it |
| `lock_timeout` | `10.0` | seconds before a lock expires (if its holder dies) |
| `lock_wait` | `5.0` | seconds a process waits for another one's value before computing it |
| `early_refresh` | `None` | fraction of the lifetime, at its end, in which hits may refresh the entry |
| `metrics` | `True` | OpenTelemetry metrics, with the `otel` extra |
| `meter_provider` | global | OpenTelemetry meter provider |
| `tracing` | `True` | OpenTelemetry spans, with the `otel` extra |
| `tracer_provider` | global | OpenTelemetry tracer provider |

### `@cache()`

| Parameter | Default | Description |
| --- | --- | --- |
| `expire` | from `init` | time to live, in seconds |
| `coder` | from `init` | |
| `key_builder` | from `init` | |
| `namespace` | `""` | group of entries, for `FastAPICache.clear(namespace=...)` |
| `injected_dependency_namespace` | `"__fastapi_alcazaba"` | prefix of the `Request`/`Response` parameters added to the endpoint |
| `exclude_params` | `None` | arguments left out of the key (a nonce, a database session…) |
| `vary_headers` | from `init` | `()` for responses that are the same for everybody |
| `respect_cache_control` | from `init` | |
| `stale_while_revalidate` | `None` | adds that `Cache-Control` directive |
| `single_flight` | from `init` | |
| `max_value_size` | from `init` | |
| `distributed_lock` | from `init` | |
| `early_refresh` | from `init` | |
| `tags` | `None` | names to invalidate the entry by: a list, or a function of the arguments |

Delete entries with `await FastAPICache.clear(namespace="products")`, one entry with
`await FastAPICache.clear(key=...)`, or by tag with `await FastAPICache.invalidate_tags(...)`.
`clear()` without arguments deletes every cached response under the prefix. Locks, rate-limit
counters, revoked tokens, one-time codes and idempotency records are kept apart, under
`{prefix}!` instead of `{prefix}:`, so no `clear()` frees a held lock, resets a limit or takes a
revocation back; `InMemoryBackend` keeps them apart too, so caching many responses never evicts
them.

<!-- --8<-- [end:configuration] -->

## Several caches, dependency injection and `get_or_set`
<!-- --8<-- [start:several-caches-dependency-injection-and-get_or_set] -->

`FastAPICache` is one global cache. For one cache per app, or several with different settings,
create `Cache` instances (same arguments as `FastAPICache.init()`) and decorate with their
`cached()`:

```python
from fastapi_alcazaba import Cache, get_cache

public_cache = Cache(RedisBackend(redis), prefix="public", vary_headers=())
private_cache = Cache(
    RedisBackend(redis), prefix="private", encryption_key=settings.cache_encryption_key
)
private_cache.install(app)  # what get_cache returns


@app.get("/catalog")
@public_cache.cached(expire=600)
async def catalog() -> list[Product]: ...


@app.get("/me/orders")
async def orders(
    user: Annotated[User, Depends(current_user)], cache: Annotated[Cache, Depends(get_cache)]
) -> list[Order]:
    # Anything worth caching that doesn't fit a decorator; the key is hashed before storing.
    return await cache.get_or_set(f"orders:{user.id}", lambda: load_orders(user.id), expire=60)
```

`get_or_set` goes through the same pipeline as the decorator (signing or encryption,
compression, single flight, locks, tags). Close the backend at shutdown with `await
cache.close()`.

<!-- --8<-- [end:several-caches-dependency-injection-and-get_or_set] -->

## Invalidating by tag
<!-- --8<-- [start:invalidating-by-tag] -->

Tag entries when they're cached and delete them all when the data changes, without knowing their
keys:

```python
@app.get("/products/{product_id}")
@cache(expire=600, tags=lambda product_id, **_: [f"product:{product_id}", "catalog"])
async def product(product_id: int) -> Product: ...


@app.put("/products/{product_id}")
async def update_product(product_id: int, data: Product) -> Product:
    saved = save(product_id, data)
    await FastAPICache.invalidate_tags(f"product:{product_id}")
    return saved
```

The function receives the same arguments as the endpoint. Tags work with the in-memory, Redis,
Valkey and GLIDE backends; in Redis-like servers each tag is a set that lives as long as its
longest-lived entry.

<!-- --8<-- [end:invalidating-by-tag] -->

## Rate limiting
<!-- --8<-- [start:rate-limiting] -->

`FastAPIRateLimiter` is a FastAPI dependency that lets each client make `times` requests per
window to a route; the next ones get a `429 Too Many Requests` with `Retry-After`. Its counters
live in the cache's backend, so every process and container shares them:

```python
from functools import partial

from fastapi import APIRouter, Depends
from starlette.requests import HTTPConnection

from fastapi_alcazaba.rate_limit import FastAPIRateLimiter


@app.get("/search", dependencies=[Depends(FastAPIRateLimiter(times=10, minutes=1))])
async def search(q: str) -> list[Product]: ...


def user_id(connection: HTTPConnection) -> str:  # a Request, or a WebSocket
    return connection.state.user_id  # set by your authentication middleware


# 1000 requests per hour and user, for all the routes of the router together.
api = APIRouter(
    dependencies=[Depends(FastAPIRateLimiter(1000, hours=1, name="api", identifier=user_id))]
)

# One budget across routes, of which each export takes 50.
partner_budget = partial(FastAPIRateLimiter, 1000, hours=1, name="partners", identifier=user_id)
partners = APIRouter()
partners.get("/orders", dependencies=[Depends(partner_budget())])(list_orders)
partners.get("/export", dependencies=[Depends(partner_budget(cost=50))])(export_orders)
```

- **Who:** by default, the client's address as the ASGI server reports it (`client_ip`), IPv6
  addresses by their /64 network: each customer gets at least a whole /64, and one budget per
  address would give a client 2⁶⁴ of them. Many ISPs delegate a /56 or a /48, and tunnel brokers
  hand out /48s for free: `client_network(24, 48)` groups wider networks. Behind
  a proxy, let the server take the address from the proxy's headers: uvicorn reads them, but
  only from the addresses in `--forwarded-allow-ips` (127.0.0.1 by default), so give it your
  proxy's. The limiter never reads `X-Forwarded-For` itself, which any client can set.
  `identifier=` counts per user, API key… Identities are hashed before they reach the backend.
- **Where:** each route has its own budget, whatever its path parameters: `/items/1` and
  `/items/2` share one. Limiters with the same `name` and rate share one budget across routes,
  and `cost=` makes a route take more of it.
- **How:** a sliding window. The count of the previous window is added in proportion to how much
  of it is still less than a window ago, so nobody doubles the limit around the boundary between
  two windows. Requests over the limit don't count: checking and counting is one atomic
  operation (a Lua script on Redis, Valkey and GLIDE, a conditional write on DynamoDB). A client
  with nothing left is then refused by each process without asking the backend, until something
  fits again: a flood doesn't reach the backend. `limiter.forget()` drops that memory (between
  tests; a new cache drops it too). Windows follow the clocks of the app's servers: keep them in
  sync (NTP).
- **What clients see:** responses carry the `RateLimit` and `RateLimit-Policy` headers of
  the [IETF draft](https://datatracker.ietf.org/doc/draft-ietf-httpapi-ratelimit-headers/):
  `RateLimit: "api";r=998;t=3540` means 998 requests left, at least for the next 3540 seconds.
  Clients can slow down before they get a 429. `headers=False` leaves them out. An endpoint that
  returns a `Response` object itself gets none: FastAPI only adds the headers that dependencies
  set to the responses it builds from a return value. Nor do the errors it builds from an
  exception (a 422, an `HTTPException` raised after the limiter), unless `FastAPIProblemDetails`
  is installed: it adds them back. An invalid request counts too: FastAPI runs the dependencies
  before it answers 422.
- **Which backend:** the `cache=` given, else the app's (`Cache.install`) or `FastAPICache`'s. Every
  backend can keep counters; `TieredBackend` keeps them only in its shared backend, and
  `FastAPICache.clear()` never resets them. They go through the cache's timeout and circuit
  breaker: if the backend fails, requests go through (`fail_open=False` answers 503 instead).
- **What happens over the limit:** `callback(request, seconds_to_wait)`, by default
  `too_many_requests` (the 429). If the callback returns, the request goes on.

On a WebSocket route, the dependency limits connections (closing with 1008). Call
`await limiter.hit(connection, key=..., cost=...)` yourself to limit anything else: it counts and
returns `None` within the limit, else the seconds to wait. For example, the messages of a WebSocket, or
the login attempts to an account:

```python
from fastapi_alcazaba.rate_limit import FastAPIRateLimiter, too_many_requests

login_attempts = FastAPIRateLimiter(times=5, minutes=15, identifier=lambda _: "")


@app.post("/login")
async def login(request: Request, form: Annotated[LoginForm, Form()]) -> Token:
    # Five attempts per account every 15 minutes, wherever they come from.
    if (wait := await login_attempts.hit(request, key=form.username)) is not None:
        too_many_requests(request, wait)
    ...
```

<!-- --8<-- [end:rate-limiting] -->

## Locks
<!-- --8<-- [start:locks] -->

A lock shared by every process and container, for work that only one of them must do at a time:

```python
from fastapi_alcazaba.background import LockNotAcquired


async def nightly_report() -> None:  # every instance runs it; one builds the report
    try:
        async with FastAPICache.lock("nightly-report", ttl=60):  # or cache.lock(...)
            await build_report()
    except LockNotAcquired:
        pass  # another instance is on it
```

- **Expiry and renewal:** a lock expires after `ttl` seconds, so a holder that dies doesn't keep it
  forever; while it's held, a background task renews it every `ttl / 3`.
- **Only the holder** releases or renews it: each acquisition has a random token, checked by a Lua
  script on Redis, Valkey and GLIDE and by conditional writes on DynamoDB (in memory, within one
  process). memcached can't hold locks. On DynamoDB, expiry compares the clocks of the app's
  servers: keep them in sync (NTP). A custom backend written for 0.1 has no `extend_lock`: use
  `renew=False` there, with a `ttl` longer than the work.
- **Waiting:** `wait=0` (the default) tries once, `wait=5` keeps trying for 5 seconds,
  `wait=None` as long as it takes. `acquire()` and `release()` work without `async with`.
- **Failures:** if the backend fails, `acquire()` raises `BackendUnavailable`: without it nobody can
  tell who holds the lock.
- **Limits:** a holder that stalls for longer than `ttl` (a long pause, a network partition) loses
  the lock while it still runs. `lock.held` turns False when the renewals couldn't keep up: check it
  between the steps of long jobs, and make work that must never overlap idempotent too.

<!-- --8<-- [end:locks] -->

## Idempotency keys
<!-- --8<-- [start:idempotency-keys] -->

A client that retries a POST after a timeout or a dropped connection gets the first response
again, instead of creating a second order or paying twice
([draft-ietf-httpapi-idempotency-key-header](https://datatracker.ietf.org/doc/draft-ietf-httpapi-idempotency-key-header/)):

```python
from fastapi_alcazaba.idempotency import IdempotencyMiddleware, require_idempotency_key

app.add_middleware(IdempotencyMiddleware)  # the app's cache (Cache.install) or FastAPICache's


@app.post("/payments", status_code=201, dependencies=[Depends(require_idempotency_key)])
async def pay(payment: Payment) -> Receipt: ...
```

The client sends a fresh key with each new request, and the same one when it retries it:
`Idempotency-Key: "8e03978e-40d5-43e8-bc93-6894a57f9324"`.

- **Retries get the first response** (status, headers and body) with `Idempotent-Replayed: true`;
  the endpoint runs once.
- **One at a time:** a retry that arrives while the first request still runs gets a 409 with
  `Retry-After: 1`. A lock in the backend, renewed while the request runs, sees to it across
  processes.
- **The same key for another request** (method, path, query or body) gets a 422. The body is
  compared byte for byte: a retry sends the same bytes, not the same JSON written another way
  (keys in another order, other spacing).
- **Per client:** keys are looked up among the requests with the same `Authorization` header
  (hashed), so one user never gets another's response. With cookie sessions, and with access
  tokens that get refreshed (a retry sent with a fresh token would run again), pass
  `identifier=` returning the user's id. Requests without credentials share one space of keys,
  safe as long as clients make their keys random (UUIDs).
- **What's kept, and for how long:** responses below 500 for `expire` seconds (24 hours), except
  those worth retrying (408, 409, 425, 429) and those about the request itself, given before the
  endpoint did anything (401, 403, 404, 405, 413, 415, 422 and the like): a retry gets them
  afresh, and nobody can fill the backend with them (FastAPI's 422 echoes the request body).
  Request bodies are read whole to fingerprint them, up to `max_body_size` (a 413 beyond); a
  client that goes away before sending all of it doesn't reach the endpoint. Responses above
  `max_response_size` aren't kept, so a retry runs again. The stored responses go through the
  cache's signing or encryption: use an `encryption_key` when they carry personal data.
- **Failures:** if the backend fails the request gets a 503: running a payment twice is worse
  than refusing it. `fail_open=True` runs it without idempotency instead. Errors of the endpoint
  itself, `BackendUnavailable` included, are left to the app.
- Requests without the header go through untouched. It applies to POST and PATCH (`methods=`),
  and needs a backend with locks.

<!-- --8<-- [end:idempotency-keys] -->

## Authentication strategies
<!-- --8<-- [start:authentication-strategies] -->

`FastAPIAuth` tells who makes a request by strategies: each strategy knows one way (a JWT, a
session, a login with GitHub…), and a dependency tries those of a route in order. The first one that finds a user gives it to the route; when none
does, a 401. Each strategy has its extra:

```shell
pip install "fastapi-alcazaba[auth-strategy-jwt]"          # JWTStrategy, with PyJWT
pip install "fastapi-alcazaba[auth-strategy-oauth2]"       # OAuth2Strategy, with httpx2
pip install "fastapi-alcazaba[auth-strategy-saml]"         # SAMLStrategy, with signxml
pip install "fastapi-alcazaba[auth-strategy-http-bearer]"  # HTTPBearerStrategy: nothing more
```

```python
from fastapi_alcazaba.auth import FastAPIAuth, SessionStrategy
from fastapi_alcazaba.auth.strategies.jwt import JWTStrategy
from fastapi_alcazaba.auth.strategies.oauth2 import OAuth2Strategy

auth = FastAPIAuth()
github = auth.use(
    OAuth2Strategy(
        name="github",
        authorization_url="https://github.com/login/oauth/authorize",
        token_url="https://github.com/login/oauth/access_token",
        userinfo_url="https://api.github.com/user",
        client_id=settings.github_client_id,
        client_secret=settings.github_client_secret,
        redirect_uri="https://example.com/auth/github/callback",
        verify=find_or_create_user,  # (tokens, profile) -> your user
    )
)
sessions = auth.use(SessionStrategy(load_user))  # load_user(user_id) -> your user
tokens = auth.use(JWTStrategy(settings.jwt_secret, verify=lambda c: load_user(c["sub"])))

CurrentUser = Annotated[User, Depends(auth.authenticate("session", "jwt"))]
GitHubUser = Annotated[User, Depends(auth.authenticate("github"))]


@app.get("/me")
async def me(user: CurrentUser) -> User:  # a browser's session, or a script's token
    return user


@app.get("/login/github")
async def login() -> RedirectResponse:
    return github.redirect()  # to GitHub, which sends the user back to the callback


@app.get("/auth/github/callback")
async def callback(request: Request, user: GitHubUser) -> RedirectResponse:
    sessions.login(request, user.id)
    return RedirectResponse("/me", status_code=303)
```

The strategies of a dependency are tried in order. Each returns the user; None when the request
doesn't carry its credentials; or raises `AuthenticationFailed` when they're there but wrong
(logged, never sent). Either way the next one is tried. When none finds a user, a 401 with the
`WWW-Authenticate` challenges of those tried, a WebSocket is closed with 1008, and
`optional=True` gives None. Any other error is a 500, never a request let through. Without a
dependency: `await auth.user(connection)`. The `verify` and `load_user` functions that turn
credentials into your user can be async or not, and return None to refuse them.

<!-- --8<-- [end:authentication-strategies] -->

### JWTStrategy
<!-- --8<-- [start:jwt-strategy] -->

`JWTStrategy` reads the token of `Authorization: Bearer` (or of the cookie named by `cookie=`)
and verifies it with PyJWT: the signature, `exp`, `nbf` and the `required` claims (`exp`, `sub`),
and `aud` and `iss` when given; a token with an `aud` is refused when no `audience` is set, as
it's meant for another service. `alg: none` never, nor HMAC and asymmetric algorithms together: a
public key taken for an HMAC secret would let anyone sign.

```python
from fastapi_alcazaba.auth import TokenBlocklist
from fastapi_alcazaba.auth.strategies.jwt import JWTStrategy

# The app's own tokens, signed with the first secret; the second one still verifies.
tokens = auth.use(
    JWTStrategy(
        [settings.jwt_secret, settings.previous_jwt_secret],
        verify=lambda claims: load_user(claims["sub"]),
        blocklist=TokenBlocklist(),
    )
)
token = tokens.issue(str(user.id), expires_in=900)

# An identity provider's tokens, with its public keys.
idp = auth.use(
    JWTStrategy(
        jwks_url="https://idp.example.com/.well-known/jwks.json",
        algorithms=["RS256"],
        audience="https://api.example.com",
        issuer="https://idp.example.com/",
        name="idp",
    )
)
```

- **Keys:** HMAC secrets have at least 32 bytes and rotate with a list; public keys need
  `algorithms=["RS256"]` (or ES256, EdDSA…), and an identity provider's keys come from its JWKS
  (`jwks_url=`, by each token's `kid`), with the `audience` and `issuer` it requires: a provider
  signs the tokens of every app it serves with the same keys.
- **Revoked tokens:** `blocklist=` refuses the tokens revoked in a `TokenBlocklist` (see "Revoked
  tokens and one-time codes").
- **Issuing:** `issue(subject, expires_in=900)` signs one, with `iat`, `exp` and a random `jti`.
- **OpenAPI:** it shows in the schema, for the docs' Authorize button.

<!-- --8<-- [end:jwt-strategy] -->

### OAuth2Strategy
<!-- --8<-- [start:oauth2-strategy] -->

`OAuth2Strategy` logs people in with the authorization code flow and PKCE, for any OAuth2 or
OpenID Connect provider (GitHub, Google, Keycloak…):

- **The login:** `redirect()` sends the browser to the provider with a random `state`; the state
  and the PKCE verifier wait in a cookie for 10 minutes (`__Host-`, HttpOnly, SameSite=Lax).
- **The callback:** the strategy checks the state, so that nobody can log a user in with their
  own account (login CSRF); exchanges the code, with the app's credentials by HTTP Basic
  (`client_auth="post"` in the form); reads `userinfo_url`; and calls `verify(tokens, profile)`.
  It only goes on the callback route: elsewhere it finds nobody.
- **The provider:** its URLs are HTTPS (or `http://localhost`); a provider that doesn't answer is
  a 502.

<!-- --8<-- [end:oauth2-strategy] -->

### SessionStrategy
<!-- --8<-- [start:session-strategy] -->

`SessionStrategy` finds the user id that `sessions.login(request, user_id)` kept in the session,
and loads the user with `load_user`. With `ServerSessionMiddleware` the session gets a new id
at login (no session fixation), and `sessions.logout(request)` deletes it from the backend. A
session whose user doesn't exist any more fails.

<!-- --8<-- [end:session-strategy] -->

### HTTPBearerStrategy
<!-- --8<-- [start:http-bearer-strategy] -->

`HTTPBearerStrategy` takes opaque tokens of `Authorization: Bearer`, which your `verify(token)`
looks up (in a database, at the provider's introspection endpoint). It needs no other library.

```python
from fastapi_alcazaba.auth import HTTPBearerStrategy


async def find_token(token: str) -> User | None:
    digest = hashlib.sha256(token.encode()).hexdigest()  # store digests, not tokens
    return await db.user_by_token_digest(digest)


bearer = auth.use(HTTPBearerStrategy(find_token))
```

A token that isn't RFC 6750's syntax, or longer than 4 KiB, fails without reaching `verify`.
Keep digests of the tokens, not the tokens. Behind a JWTStrategy, JWTs go to that one and opaque
tokens to this one.

<!-- --8<-- [end:http-bearer-strategy] -->

### SAMLStrategy
<!-- --8<-- [start:saml-strategy] -->

`SAMLStrategy` is SAML 2.0 single sign-on with an identity provider (Okta, Entra ID, ADFS,
Keycloak…):

- **The login:** `redirect()` sends the browser to the IdP with an AuthnRequest; the IdP posts its
  answer to the ACS route.
- **The answer:** the strategy verifies its XML signature with the IdP's certificate (signxml,
  SHA-2 only, no DTDs). Then, on what the signature covers and nothing else (signature
  wrapping), it checks the issuer, the audience (`entity_id`), the recipient (`acs_url`), the
  request it answers and when it's valid.
- **Once:** each assertion works once: it's kept in the cache's backend until it expires, and
  logins are refused when the backend fails.
- **Metadata:** `metadata()` is the app's metadata, for the IdP.
- **Not supported:** logins the IdP starts, encrypted assertions, signed requests, single logout.

The SAML request's id waits in a `SameSite=None` cookie, because the IdP posts the answer from
its own site; for the same reason `CSRFMiddleware` must let the ACS through:

```python
from fastapi_alcazaba.auth.strategies.saml import SAMLStrategy

saml = auth.use(
    SAMLStrategy(
        entity_id="https://example.com/saml/metadata",  # this app, at the IdP
        acs_url="https://example.com/saml/acs",
        idp_entity_id=settings.idp_entity_id,
        idp_sso_url=settings.idp_sso_url,
        idp_certificate=settings.idp_certificate,  # PEM; a list while it rotates
        verify=find_or_create_user,  # (SAMLProfile) -> your user
    )
)
SAMLUser = Annotated[User, Depends(auth.authenticate("saml"))]
app.add_middleware(CSRFMiddleware, exempt_paths=["/saml/acs"])  # the IdP posts it


@app.get("/saml/login")
async def saml_login() -> RedirectResponse:
    return saml.redirect()


@app.post("/saml/acs")
async def saml_acs(request: Request, user: SAMLUser) -> RedirectResponse:
    sessions.login(request, user.id)
    return RedirectResponse("/me", status_code=303)


@app.get("/saml/metadata")
async def saml_metadata() -> Response:
    return Response(saml.metadata(), media_type="application/samlmetadata+xml")
```

<!-- --8<-- [end:saml-strategy] -->

### Your own strategy
<!-- --8<-- [start:your-own-strategy] -->

Your own strategy (an API key, a client certificate, a signed request…) is a `Strategy` whose
`authenticate()` returns the user, None, or raises `AuthenticationFailed`; `challenge` and
`openapi_scheme()` are optional:

```python
from fastapi import HTTPException
from starlette.requests import HTTPConnection

from fastapi_alcazaba.auth import AuthenticationFailed, FastAPIAPIKeys, Strategy


class APIKeyStrategy(Strategy):
    challenge = "APIKey"

    def __init__(self, keys: FastAPIAPIKeys[Any]) -> None:
        super().__init__("api-key")
        self.keys = keys

    async def authenticate(self, connection: HTTPConnection) -> Any:
        key = connection.headers.get("x-api-key")
        if key is None:
            return None  # not this strategy's: the next one is tried
        try:
            return await self.keys.authenticate(key)
        except HTTPException as exc:  # FastAPIAPIKeys' 401
            raise AuthenticationFailed("invalid API key") from exc


auth.use(APIKeyStrategy(keys))
```

<!-- --8<-- [end:your-own-strategy] -->

## Revoked tokens and one-time codes
<!-- --8<-- [start:revoked-tokens-and-one-time-codes] -->

A JWT can't be taken back once issued. `TokenBlocklist` keeps the revoked ones in the cache's
backend, shared by every process, until they would have expired anyway:

```python
from fastapi_alcazaba.auth import OneTimeCodes, TokenBlocklist

blocklist = TokenBlocklist()  # FastAPICache's backend; or TokenBlocklist(cache)


async def current_user(claims: Annotated[dict[str, Any], Depends(decode_jwt)]) -> str:
    sub, iat = claims["sub"], claims["iat"]
    if await blocklist.is_revoked(claims["jti"], subject=sub, issued_at=iat):
        raise HTTPException(401)
    return sub


@app.post("/logout")
async def logout(claims: Annotated[dict[str, Any], Depends(decode_jwt)]) -> None:
    await blocklist.revoke(claims["jti"], expires_at=claims["exp"])


@app.post("/password")
async def change_password(user: Annotated[str, Depends(current_user)]) -> Token:
    ...
    await blocklist.revoke_subject(user, ttl=TOKEN_LIFETIME)  # log out everywhere
    return issue_token(user)  # issued now: still valid
```

`OneTimeCodes` are the codes you send by e-mail or SMS:

```python
codes = OneTimeCodes(secret=settings.otp_secret)  # secrets.token_urlsafe(32)


@app.post("/login/code", dependencies=[Depends(FastAPIRateLimiter(times=3, minutes=15))])
async def send_code(email: str) -> None:
    await send_email(email, await codes.issue(email, purpose="login"))


@app.post("/login")
async def login(email: str, code: str) -> Token:
    if not await codes.verify(email, code, purpose="login"):
        raise HTTPException(400, "Invalid or expired code")
    ...
```

- **Nothing in clear:** token ids and subjects are stored as SHA-256 digests. Codes are stored as
  an HMAC with `secret`, bound to their subject and purpose: a plain digest of a 6-digit code is
  brute-forced in milliseconds by whoever reads the backend. A list of secrets rotates them.
- **Expiry:** each revocation lasts while the token could still be accepted (`exp`, plus
  `leeway`, 60 s by default, for what your JWT library tolerates). Codes last `ttl` seconds (300).
- **Log out everywhere:** `revoke_subject` refuses the tokens issued before the current second.
  `iat` has whole seconds, so the token you issue right after for the current session stays
  valid.
- **Guesses and single use:** every guess counts, malformed ones included, and a code is lost
  after `max_attempts` (5). Of several requests with the right code at the same time, only one
  succeeds. Issuing a code replaces the previous one and gives the attempts again: rate-limit the
  route that issues them.
- **Every process sees them at once:** they're read from the shared backend, never from
  `TieredBackend`'s copies in each process. On DynamoDB, whose reads are eventually consistent, a
  revocation or a new code can take about a second to reach every process.
- **Failures:** if the backend fails, `is_revoked` refuses tokens (`fail_open=True` accepts them),
  `verify` returns False, and `revoke`, `revoke_subject` and `issue` raise `BackendUnavailable`.
  With a `signing_key` or `encryption_key`, revocations are sealed and planted ones are ignored.
  Whoever can delete keys in the backend can still un-revoke tokens.

<!-- --8<-- [end:revoked-tokens-and-one-time-codes] -->

## API keys
<!-- --8<-- [start:api-keys] -->

`FastAPIAPIKeys` authenticates machine clients by an API key. It generates the keys and checks
them; your database keeps a record of each, never the key itself:

```python
from fastapi_alcazaba.auth import FastAPIAPIKeys


async def load_key(key_id: str) -> ApiKeyRow | None:  # any object with digest, scopes, expires_at
    return await db.get(ApiKeyRow, key_id)


keys = FastAPIAPIKeys(load_key, pepper=settings.api_key_pepper, bearer=True)


@app.post("/api-keys")
async def create_key(user: Annotated[User, Depends(current_user)]) -> dict[str, str]:
    new = keys.generate(scopes=["orders:read"], expires_at=datetime.now(UTC) + timedelta(days=90))
    await db.add(ApiKeyRow(owner=user.id, **dataclasses.asdict(new.record)))
    return {"key": new.key}  # shown once: only its digest is stored


@app.get("/orders")
async def orders(key: Annotated[ApiKeyRow, Depends(keys.require("orders:read"))]) -> list[Order]:
    return await db.orders(owner=key.owner)
```

- **Key format:** `ak_<id>_<secret>` (`prefix` changes `ak`; `sk_live` works too): an 80-bit id
  that finds the record and a 256-bit secret, in lowercase base32. A distinctive prefix lets
  secret scanners (GitHub's, GitLab's) spot leaked keys.
- **Stored as a digest:** the record holds a SHA-256 digest of the secret, bound to the key's id.
  Random 256-bit secrets don't need a slow password hash. With a `pepper` (a server secret of at
  least 32 bytes, kept out of the database) the digest is an HMAC, so a copy of the database
  alone can't even test a guess. A list of peppers rotates them: the first one makes new keys,
  all of them are accepted.
- **Your database:** `lookup(key_id)` returns your row, async or not (a plain function runs in a
  thread, like FastAPI's sync dependencies), and the dependency returns that same row. Any object
  with `digest`, `scopes` (a collection, or a space-separated string) and `expires_at` works;
  `APIKeyRecord` is what `generate()` gives.
- **Where the key comes from:** the `X-API-Key` header (`header=...`) and, with `bearer=True`,
  `Authorization: Bearer <key>`. Both appear in the OpenAPI schema, for the docs' Authorize
  button.
- **One answer:** a missing, malformed, unknown, wrong or expired key gets the same 401, with
  `WWW-Authenticate`; the secret is compared in constant time, and malformed keys don't reach
  `lookup`. `keys.require("a", "b")` adds a 403 for a valid key without all those scopes (with an
  RFC 6750 `insufficient_scope` challenge in bearer mode). Keys and secrets are never logged.
- **Revocation:** delete the record, or set its `expires_at` in the past (naive datetimes are
  taken as UTC). `cache_ttl` keeps looked-up records in each process for that many seconds to
  spare the database, so a revocation then takes up to `cache_ttl` to reach every process;
  `keys.forget(key_id)` drops one from this process at once. Unknown ids are never cached.
- **Floods:** each request with a well-formed key costs a lookup: put a `FastAPIRateLimiter`
  in front.
- **Without a dependency:** `await keys.authenticate(key)` for keys that come some other way (a
  WebSocket message).

<!-- --8<-- [end:api-keys] -->

## Authorization with Casbin
<!-- --8<-- [start:authorization-with-casbin] -->

`FastAPICasbin` lets a request through only when a [Casbin](https://casbin.org) policy allows it.
Who may do what lives in the policy (ACL, RBAC, RBAC with domains, ABAC…), not in checks spread
over the routes. The `casbin` extra installs [PyCasbin](https://github.com/casbin/pycasbin).

A Casbin model says what a request is and how it matches the policy; the policy, who may do
what. Here a request is (user, object, action), and users get permissions through roles:

```ini title="model.conf"
[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = g(r.sub, p.sub) && keyMatch2(r.obj, p.obj) && r.act == p.act
```

```text title="policy.csv"
p, reader, /articles/:id, GET
p, editor, /articles/:id, PUT
p, editor, articles, publish
g, editor, reader
g, ana, editor
```

```python
import casbin

from fastapi_alcazaba.authz import FastAPICasbin

enforcer = casbin.Enforcer("model.conf", "policy.csv")
authz = FastAPICasbin(enforcer, subject=current_username)  # your dependency: who's asking

# (user, path, method): may ana PUT /articles/7?
api = APIRouter(dependencies=[Depends(authz)])


# (user, *values): your own objects and actions. Both dependencies return the user.
@app.post("/articles/{article_id}/publish")
async def publish(
    article_id: int,
    editor: Annotated[str, Depends(authz.require("articles", "publish"))],
) -> Article: ...


# {name} takes a path parameter: (user, "projects/7", "archive").
archive = authz.require("projects/{project_id}", "archive")


@app.post("/projects/{project_id}/archive", dependencies=[Depends(archive)])
async def archive_project(project_id: int) -> None: ...
```

- **Who:** `subject` is a FastAPI dependency that returns who makes the request: a user name, an
  id, or an object for ABAC matchers (`r.sub.age >= 18`); None when nobody is logged in. Its own
  dependencies (a bearer token, a session) appear in the OpenAPI schema, for the docs' Authorize
  button. Without one, the user of Starlette's `AuthenticationMiddleware`.
- **What:** `Depends(authz)` checks the path the app routes (without the `root_path` a proxy or
  a mount adds in front) and the method; WebSockets have the action `WEBSOCKET`, so a policy for
  GET doesn't open them. `authz.require(...)` checks your own values instead, `{name}` filled
  from the path parameters. In `keyMatch2`, `:name` is a parameter: name objects without colons
  (`projects/7`, not `project:7`), or the rule for `project:7` lets `project:8` through.
- **Answers:** a request without a subject gets a 401 with `WWW-Authenticate: Bearer`
  (`challenge=`), without asking the policy; `anonymous="anonymous"` has the policy check it as
  that subject instead, for routes open to everybody (`p, anonymous, /catalog, GET`). A subject
  that the policy refuses gets a 403. WebSockets are closed with 1008 (policy violation). A path
  with control characters (`/users/%0A`) gets a 400 without asking the policy: Starlette routes
  it to `/users/`, while the policy would read the newline. With `FastAPIProblemDetails`, the
  errors are problem details.
- **In a route:** `authz.enforce(user, "drafts", "read")` returns a bool, for what depends on the
  data: the items of a list, the owner of an object.
- **Fails closed:** a broken model, or a request that doesn't fit it (`invalid request size`),
  raises: a 500, never a request let through. A disabled enforcer
  (`enforcer.enable_enforce(False)`) lets everything through, as Casbin does.
- **Enforcers:** any of Casbin's: `Enforcer`, `SyncedEnforcer`, `FastEnforcer`, `AsyncEnforcer`
  (for async adapters; `await enforcer.load_policy()` in the lifespan). The checks run in the
  event loop, like an AsyncEnforcer's changes: a request never sees half a change. Casbin tries
  the rules one by one, a few microseconds each (about 2.5 ms per request with 1,000 rules on a
  laptop): give permissions to roles rather than users, and roles to users.
- **Several processes:** each one has its copy of the policy. A change made in one
  (`await enforcer.add_role_for_user(...)`) reaches the database through the adapter, and the
  other processes when they load the policy again: a Casbin watcher (casbin-redis-watcher…)
  tells them.

`CasbinMiddleware` checks every request instead, as (user, path, method) with the user of
Starlette's `AuthenticationMiddleware`, `anonymous` when there's none. Add it before
`AuthenticationMiddleware`, so that it runs after it; the policy must then open the public
routes (`/docs`, `/login`…) to `anonymous`:

```python
from starlette.middleware.authentication import AuthenticationMiddleware

from fastapi_alcazaba.authz import CasbinMiddleware

app.add_middleware(CasbinMiddleware, enforcer=enforcer)
app.add_middleware(AuthenticationMiddleware, backend=JWTBackend())
```

It checks the path the app routes, without the `root_path`, and the method, `WEBSOCKET` for
WebSockets. Refusals get a 403 with `{"detail": "Forbidden"}`, like FastAPI's errors (problem
details with `FastAPIProblemDetails`), or a 401 with `WWW-Authenticate` when `anonymous` is
refused, and a 400 for paths with control characters; WebSockets are closed with 1008. `OPTIONS`
requests are checked like the others: add `CORSMiddleware` after `CasbinMiddleware` (outside it),
to answer the preflights.

<!-- --8<-- [end:authorization-with-casbin] -->

## Signed webhooks
<!-- --8<-- [start:signed-webhooks] -->

`FastAPIWebhooks` accepts only webhooks signed with your secret, and each delivery once.
`WebhookSigner` signs the ones you send. Both follow
[Standard Webhooks](https://www.standardwebhooks.com), the format of Svix, Supabase, Clerk,
Resend, OpenAI and others: `webhook-id`, `webhook-timestamp` and
`webhook-signature: v1,<base64 HMAC-SHA256 of "id.timestamp.body">`, with `whsec_...` secrets.

```python
from fastapi_alcazaba.webhooks import FastAPIWebhooks, WebhookMessage, WebhookSigner

webhooks = FastAPIWebhooks(settings.webhook_secret)  # "whsec_..."


@app.post("/webhooks/payments")
async def payment_event(message: Annotated[WebhookMessage, Depends(webhooks)]) -> None:
    event = message.json()
    ...  # idempotent on message.id: retries of a message keep its id


# Sending them: sign the exact bytes you send.
signer = WebhookSigner(settings.webhook_secret)  # WebhookSigner.generate_secret() makes one
body = json.dumps(event).encode()
await client.post(url, content=body, headers=signer.sign(body))
```

- **Raw body, still readable:** the signature is checked over the body exactly as it arrived,
  read up to `max_body_size` bytes (1 MiB; a 413 beyond, and a `Content-Length` over it is
  refused before reading). It stays in the request, for `message.json()` or `await
  request.json()`. Validate it with your model there (`Event.model_validate_json(message.body)`)
  rather than with a pydantic body parameter: FastAPI reads and validates those before any
  dependency runs, so the body would be read without the limit, and an unsigned malformed one
  would get a 422 before its signature is checked. Not a form parameter either, which consumes
  the body first.
- **Freshness:** a timestamp more than `tolerance` seconds (300) away from the server's clock,
  in the past or the future, is refused.
- **Constant time, one answer:** every signature in the header is compared in constant time
  against each secret. A wrong signature, a stale timestamp and a replay all get the same 401
  (`"Invalid webhook"`); only missing or malformed headers get a 400.
- **Replays:** each accepted delivery (its id and timestamp, as a SHA-256 digest under
  `Cache.state_key`) is counted in the cache's backend until its timestamp can't be accepted
  anymore, so a captured request sent again is refused by every process. Only accepted
  deliveries are counted: a forged copy can't burn the id of the real one. A retry of the sender
  has a new timestamp and signature and goes through, as it must after a failed attempt: make the
  endpoint idempotent on `message.id` (a unique constraint) if a message must not be processed
  twice. The backend must support counters, as every built-in backend does.
- **Backend down:** a 503 with `Retry-After` (fail closed): senders retry later, while a replay
  would run its effects again. `fail_open=True` accepts correctly signed webhooks without the
  replay check instead.
- **Rotation:** `FastAPIWebhooks([new, old])` accepts a signature by either secret;
  `WebhookSigner([new, old])` signs every message with both, so receivers with either secret
  accept it.
- **Without a dependency:** `webhooks.verify(headers, body)` checks the signature and the
  timestamp (not replays), for queues and scripts.

<!-- --8<-- [end:signed-webhooks] -->

## Outbound calls
<!-- --8<-- [start:outbound-calls] -->

`ResilientTransport` is an httpx transport for the calls your app makes to other services: a
deadline for each call and a circuit breaker for each origin. It needs httpx2 (or httpx):

```python
import httpx2 as httpx

from fastapi_alcazaba.outbound import CircuitOpenError, ResilientTransport


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    transport = ResilientTransport(timeout=5, failure_threshold=5, reset_timeout=30)
    async with httpx.AsyncClient(transport=transport, base_url="https://api.example.com") as client:
        app.state.partner_api = client
        yield


@app.get("/quote")
async def quote(request: Request) -> Quote:
    try:
        response = await request.app.state.partner_api.get("/quote")
    except CircuitOpenError:  # the service keeps failing: don't wait for it
        raise HTTPException(503, "Quotes are unavailable, try again later")
    return Quote.model_validate(response.json())
```

- **Deadline:** `timeout` bounds the whole call, from sending the request to the last byte of the
  response. httpx's own timeouts are per phase, and its read timeout restarts with every byte, so
  a server that trickles a byte every few seconds would keep the request open forever. Past the
  deadline, `httpx.TimeoutException`. It applies to each request the transport sends: with
  `follow_redirects=True`, every hop gets a deadline of its own.
- **Response size:** responses are read whole before they're returned, so don't stream large
  downloads through it. `max_response_size` caps the bytes read (as received, before
  decompression); past it, `ResponseTooLargeError`, an `httpx.RequestError`.
- **Circuit breaker:** each origin (scheme, host and port) has its own. After
  `failure_threshold` consecutive failures, calls to that origin fail at once with
  `CircuitOpenError` for `reset_timeout` seconds, without reaching the server; then a single
  trial call decides whether it's back. `CircuitOpenError` is an `httpx.TransportError`, so code
  that already handles httpx errors handles it too. `transport.state(url)` says `"closed"`,
  `"open"` or `"half-open"`, handy in a health check. `failure_threshold=None` turns it off.
- **What counts as a failure:** the origin's transport errors (connection and protocol errors,
  timeouts, yours and httpx's) and responses whose status is in `failure_statuses` (500, 502, 503
  and 504 by default), which are still returned to you. Any other response, 4xx included, is a
  success. Your own errors are neither: an illegal header value (`httpx.LocalProtocolError`), a
  URL scheme httpx can't send (`httpx.UnsupportedProtocol`), an exception from your request body,
  `ResponseTooLargeError`. Otherwise anyone who can put a string in an outbound header could
  switch the integration off for everybody.
- **Your own transport settings:** `ResilientTransport(httpx.AsyncHTTPTransport(verify=...,
  limits=..., proxy=...))`.

### URLs your users give: SSRF

Calling a URL that a user typed (a webhook endpoint, a link preview, an avatar by URL) lets them
make your server call its own network: `http://169.254.169.254/` (the cloud's metadata service
and its credentials), `http://localhost:6379/`, an admin panel on `10.0.0.0/8`.
`SSRFGuardTransport` only calls the public internet:

```python
from fastapi_alcazaba.outbound import (
    BlockedDestinationError,
    ResilientTransport,
    SSRFGuardTransport,
)

client = httpx.AsyncClient(
    transport=ResilientTransport(SSRFGuardTransport(ports=(443,)), timeout=10),
    follow_redirects=True,
)

try:
    await client.post(subscription.url, content=body, headers=signer.sign(body))
except BlockedDestinationError:
    ...  # tell the user the URL isn't allowed
```

- **What's refused:** a host with any address that isn't public: private, loopback, link-local
  (169.254.0.0/16 and its metadata services, `fe80::/10`), carrier-grade NAT (100.64.0.0/10),
  multicast, reserved, documentation and benchmarking ranges, unique-local IPv6 (`fc00::/7`, AWS's
  `fd00:ec2::254`), and the IPv6 forms of those IPv4 addresses (`::ffff:127.0.0.1`, NAT64
  `64:ff9b::/96`, 6to4, Teredo). Literal addresses in URLs are checked without DNS; anything else
  goes to DNS, so numeric spellings (`http://2130706433/`, `http://0x7f.1/`) and `localhost`
  are refused for what they resolve to. Also refused: schemes other than http and https and, with
  `ports`, other ports. `BlockedDestinationError` says which host, never the address it resolved
  to (that's logged instead).
- **Pinned against DNS rebinding:** the transport resolves the host itself and connects to the
  address it checked: the URL is rewritten to it, the `Host` header keeps the name, and TLS gets
  the name for SNI and the certificate check. A DNS server that answers a public address to the
  check and a private one to the connection gets nowhere. If the first address can't be reached,
  one of the other family (IPv4/IPv6) is tried.
- **Redirects:** with `follow_redirects=True`, every hop goes through the transport and is checked
  like the first one.
- **Checking a URL when it's given:** `await guard.check(url)` checks it as a request would be
  (scheme, port, every address its host resolves to) without sending anything, so a webhook's
  endpoint can be refused when a user registers it. Each request is checked again all the same:
  DNS answers change.
- **Allowed internal destinations:** `allow=["billing.internal", "*.svc.cluster.local",
  "10.1.2.0/24"]` lets those names (exact, or subdomains) and networks through although they
  aren't public; they're still pinned.
- **With `ResilientTransport`:** wrap the guard in it (not the other way round): the deadline
  then covers the DNS lookup and each breaker is per host name. `BlockedDestinationError` is an
  `httpx.RequestError` but not a `TransportError`, so it doesn't count against the origin's
  breaker or trigger retry policies.
- **Connections:** each host name gets an `httpx.AsyncHTTPTransport` of its own (sharing one SSL
  context), at most `max_hosts` (100) kept open: pinned connections are pooled by address, and two
  names on one address (a CDN) must not share a connection whose certificate was checked for the
  other. Pass a factory for your own settings, `SSRFGuardTransport(lambda:
  httpx.AsyncHTTPTransport(http2=True))`, or a transport instance (a mock in tests), which then
  carries every request.
- **Proxies:** a proxy connects on its own, so pinning can't work through it:
  `httpx.AsyncHTTPTransport(proxy=...)` as the inner transport is refused. Don't pass `proxy=` or
  `mounts=` to the client either: those requests would bypass the transport. (The client ignores
  the `HTTP_PROXY` variables when it's given a transport.) An egress proxy that filters
  destinations itself, such as Smokescreen, is the alternative.

<!-- --8<-- [end:outbound-calls] -->

## Security headers and body size
<!-- --8<-- [start:security-headers-and-body-size] -->

Two ASGI middlewares for what FastAPI leaves to the deployment:

```python
from fastapi_alcazaba.security import BodySizeLimitMiddleware, SecurityHeadersMiddleware

app.add_middleware(
    BodySizeLimitMiddleware,
    max_size=1_048_576,  # 1 MiB
    paths={"/uploads": 50 * 1_048_576, "/internal/stream": None},
)
app.add_middleware(SecurityHeadersMiddleware)  # last: the outermost
```

- **Security headers:** the OWASP REST ones: `X-Content-Type-Options: nosniff`,
  `X-Frame-Options: DENY`, `Referrer-Policy: no-referrer` and `Cross-Origin-Opener-Policy:
  same-origin`. Headers the app sets itself are never overwritten; `headers=` adds others or
  changes the defaults (None removes one). Add it last, the outermost, so that the answers of
  the middlewares inside it (`BodySizeLimitMiddleware`'s 400 and 413) get them too. Starlette's
  500 for an unhandled exception never does: `ServerErrorMiddleware` wraps every middleware you
  add.
- **HSTS** only on HTTPS requests: browsers ignore it over plain HTTP, and on a development
  `localhost` it would stick. Behind a proxy that terminates TLS, let the server trust it
  (`uvicorn --proxy-headers --forwarded-allow-ips=<proxy address>`).
- **Content-Security-Policy:** `default-src 'none'; frame-ancestors 'none'` for everything that
  isn't HTML (nothing in a JSON response should load anything). HTML responses get `html_csp`,
  None by default, so that the Swagger UI and ReDoc, which load scripts from a CDN, keep working.
- **Body size:** FastAPI has no limit of its own; a client could send an endless body that the app
  reads into memory. A `Content-Length` over the limit gets a 413 before the app runs, a malformed
  or ambiguous one a 400. Bodies without one (chunked) are counted while the app reads them and
  refused once they pass the limit, also when FastAPI parses them into a model. `paths` sets
  other limits by path prefix (the longest wins, None for no limit), matched like the routes,
  without the `root_path`.
- Both are plain ASGI middlewares, not `BaseHTTPMiddleware`: streaming responses still stream,
  and WebSockets go through untouched.

<!-- --8<-- [end:security-headers-and-body-size] -->

## CSRF protection
<!-- --8<-- [start:csrf-protection] -->

A page on another site can make the user's browser send a form, or a `fetch`, to your app, and
the browser adds the user's cookies by itself. `CSRFMiddleware` refuses those requests with a 403,
from the headers browsers send to say where a request comes from:

```python
from fastapi_alcazaba.security import CSRFMiddleware

app.add_middleware(
    CSRFMiddleware,
    trusted_origins=["https://admin.example.com"],  # other origins whose pages may post here
    exempt_paths=["/webhooks"],  # signed in their own way
)
```

- **Fetch metadata first:** every current browser sends `Sec-Fetch-Site`. `same-origin` (and
  `none`, a request the user typed) goes through; `same-site` and `cross-site` are refused unless
  their `Origin` is in `trusted_origins`. Sibling subdomains are other origins too.
- **Older browsers:** without `Sec-Fetch-Site`, the `Origin` header (or else the `Referer`) must
  be the app's own origin or a trusted one. `Origin: null` (sandboxed frames, redirects) is
  refused, and so is a request with neither header.
- **Only what the browser adds by itself is protected:** requests without cookies or remembered
  HTTP authentication (Basic, Digest, Negotiate, NTLM) go through, so APIs that take a bearer
  token in a header keep working from anywhere (a page on another site can't set the header).
  Except form submissions from other sites (`Sec-Fetch-Mode` other than `cors`): before the user
  has a cookie they would log the browser into the attacker's account (login CSRF).
  `check_all=True` checks every request, for apps that trust TLS client certificates or the
  network.
- **Safe methods** (GET, HEAD, OPTIONS, TRACE) always go through: they mustn't change anything.
- **The app's own origin** is the scheme and the `Host` header as the server sees them. Behind a
  proxy that terminates TLS, let the server trust it (`uvicorn --proxy-headers
  --forwarded-allow-ips=<proxy address>`) and have the proxy pass `Host` on; otherwise every
  request looks cross-site, so list the public origin in `trusted_origins`.
- **Double-submit token** for single-page apps, as defence in depth: with `require_token=True`
  every response to a request without a valid token sets a random one (256 bits) in the
  `__Host-csrftoken` cookie (readable by JavaScript, `Secure`, `SameSite=Lax`, the `__Host-`
  prefix keeps sibling subdomains from planting one), also in `request.state.csrf_token`. Requests
  with cookies must send it back in `X-CSRF-Token`; it's compared in constant time. A browser that
  sends neither Origin nor Referer passes with a valid token. HTML forms can't set headers, so
  with `require_token=True` every form that changes something must be sent by JavaScript (`fetch`
  with the header), the login form included; apps with plain HTML forms leave it off and rely on
  the origin checks, which protect them on their own.
- **WebSockets:** browsers don't apply CORS to WebSocket handshakes, so any site could open one
  with the user's cookies and talk to it (cross-site WebSocket hijacking). Handshakes with cookies
  get the same checks and are closed with 1008 before they're accepted (the server answers 403);
  handshakes without `Origin` aren't from a browser and go through.
- Refusals are logged at INFO with the method and the reason (never cookies or tokens), and
  counted in `fastapi_alcazaba.csrf.refused`. A plain ASGI middleware: streaming responses still
  stream.

<!-- --8<-- [end:csrf-protection] -->

## Server-side sessions
<!-- --8<-- [start:server-side-sessions] -->

Starlette's `SessionMiddleware` keeps the whole session in a signed cookie: the client can read
it (it's only base64), it can't be revoked (a copied cookie keeps working after logout, until it
expires) and it can't hold much more than 4 KB. `ServerSessionMiddleware` keeps the data in the
cache's backend instead, and the cookie only carries a random id; `request.session` works the
same:

```python
from fastapi_alcazaba.security import ServerSessionMiddleware, clear_session, rotate_session

app.add_middleware(ServerSessionMiddleware, max_age=8 * 3600, idle_timeout=1800)


@app.post("/login")
async def login(request: Request, form: Annotated[Login, Form()]) -> RedirectResponse:
    user = await authenticate(form)
    request.session["user_id"] = user.id
    rotate_session(request)  # a new id: one planted in the browser before is now worthless
    return RedirectResponse("/", status_code=303)


@app.post("/logout")
async def logout(request: Request) -> RedirectResponse:
    clear_session(request)  # deleted from the backend: a copied cookie is worth nothing now
    return RedirectResponse("/", status_code=303)
```

- **The cookie** holds 256 random bits (`secrets.token_urlsafe(32)`), `HttpOnly`, `Secure` and
  `SameSite=Lax` by default, with `Max-Age` = `max_age`. Use `cookie_name="__Host-session"` so that
  a sibling subdomain can't plant one (cookie tossing).
- **The backend** keeps the data under a SHA-256 digest of the id (whoever reads the keys can't use
  them as cookies), sealed with the cache's `signing_key` or `encryption_key` (use the latter when
  sessions hold personal data). Without either, whoever can write to the backend can forge a
  session, and a warning is logged. With a `TieredBackend` only the shared tier is used, so a
  logout is seen by every process at once.
- **Saved only when it changed**, nested changes included (`request.session["cart"].append(...)`),
  when the response starts. Empty sessions aren't stored and get no cookie; a session the app
  empties is deleted and its cookie expired. Responses that read it get `Vary: Cookie`, and those
  that set the cookie `Cache-Control: no-store` unless they set their own.
- **New ids are always the server's:** a cookie whose session doesn't exist (expired, made up)
  never becomes the id of a new session. `rotate_session` gives a new id with the same data (and
  a fresh `max_age`) at login, against session fixation; `clear_session` deletes the session and
  expires the cookie at logout, and data stored afterwards goes to a new session.
- **Expiry:** `max_age` seconds after creation or rotation, and `idle_timeout` seconds without
  requests (refreshed in the backend at most every tenth of it, not on every request). Both are
  checked when the session is read, and the backend drops it then as well.
- **JSON only**, up to `max_size` bytes (16 KiB): more, or a value that isn't JSON, raises when
  the response starts.
- **Backend failures:** if the session can't be read, the request goes on with an empty one
  (anonymous, which fails closed) and a warning. Changing it then, or failing to save, rotate or
  clear it, replaces the response with a 503 and leaves the cookie alone: the response would claim
  a login or a logout that didn't happen. Concurrent requests of one session: the last save wins.
- WebSockets get `websocket.session` too, read-only (`rotate_session` and `clear_session` raise
  there). A plain ASGI middleware: streaming responses still stream.

<!-- --8<-- [end:server-side-sessions] -->

## Checking uploads
<!-- --8<-- [start:checking-uploads] -->

The name and the `Content-Type` of an upload are whatever the client chose: an HTML page sent as
`cat.png` would be served back as HTML. `FastAPIUploads` checks the file itself, with nothing but
the standard library:

```python
from fastapi_alcazaba.uploads import IMAGES, CheckedFile, FastAPIUploads

uploads = FastAPIUploads(types=IMAGES, max_size=5 * 1024 * 1024, max_pixels=25_000_000)


@app.post("/avatars")
async def avatar(upload: Annotated[CheckedFile, Depends(uploads)]) -> dict[str, str]:
    name = f"{uuid4()}{upload.extension}"  # never the client's name
    await storage.save(name, upload.file.file, content_type=upload.content_type)
    return {"name": name}
```

As a dependency it reads the form field `file`; for other fields or several files, call
`await uploads.check(file)`. It returns a `CheckedFile`: the `UploadFile` read back to its start,
the `content_type` found in its bytes, its `size`, its `width` and `height` (images) and a safe
`filename`.

- **The real type** (`sniff(data)`): PNG, JPEG, GIF, WebP, PDF, ZIP, GZIP, MP4 and QuickTime from
  their signatures; Word, Excel and PowerPoint documents from the content types their
  `[Content_Types].xml` declares (as Office reads them), with the macro-enabled ones (`.docm`,
  `.xlsm`, `.pptm`) told apart, and anything ambiguous reported as a plain ZIP; UTF-8 text without
  control characters (`text/plain`: CSV is text, it can't be told apart). Text that starts with a
  tag is HTML, SVG or XML, never plain text, and SVG is never an image: all three can carry
  scripts. `types` must come from `KNOWN_TYPES` (`IMAGES`, `OFFICE_DOCUMENTS`...), so a typo like
  `image/jpg` fails at start-up. Store and serve files with the detected type, never the
  client's, plus `X-Content-Type-Options: nosniff` (`SecurityHeadersMiddleware`).
- **Size:** counted while the file is read in chunks, whatever `UploadFile.size` says; 413 beyond
  `max_size`, 415 for a type that isn't allowed, 422 for an empty file. By the time an endpoint
  runs the multipart parser has stored the whole request already, so limit it with
  `BodySizeLimitMiddleware` on the route as well.
- **Decompression bombs:** the width and height of PNG, JPEG (however much metadata comes before
  the frame), GIF and WebP images are read from their headers (`image_size(data)`) without
  decoding them; more than `max_pixels` (40 million by default) gets a 413, and an image whose
  dimensions can't be read a 415.
- **Safe names** (`safe_filename(name)`): NFKC-normalised, only the last path component, without
  control or invisible characters (such as the right-to-left override that shows `exe.txt` as
  `txt.exe`), `<>:"|?*` replaced, without leading dots or dashes, `_` in front of the names Windows
  reserves (`CON`, `NUL`, `COM1`...), and at most 255 bytes, keeping the extension. For display and
  `Content-Disposition`; store files under names of your own (`CheckedFile.extension`).
- **What it doesn't do:** look inside the files. A PDF can carry scripts, Office documents
  macros, archives may expand to far more than their size, and the frames of a GIF aren't counted.
  Allow HTML, SVG or XML only to serve them as attachments from another domain.

<!-- --8<-- [end:checking-uploads] -->

## Problem details
<!-- --8<-- [start:problem-details] -->

`FastAPIProblemDetails` answers errors with RFC 9457 problem details
(`application/problem+json`) instead of FastAPI's `{"detail": ...}`, and without telling clients
what they shouldn't know:

```python
from fastapi_alcazaba.observability import FastAPIProblemDetails, Problem

problems = FastAPIProblemDetails()
problems.install(app)


@app.post("/orders")
async def order(item: Item) -> Order:
    if stock[item.id] < item.quantity:
        raise Problem(
            409,
            f"Only {stock[item.id]} left",
            type="https://example.com/problems/out-of-stock",
            title="Out of stock",
            extensions={"available": stock[item.id]},
        )
    ...
```

```json
{"type": "https://example.com/problems/out-of-stock", "title": "Out of stock", "status": 409,
 "detail": "Only 3 left", "instance": "/orders", "available": 3, "request_id": "4bf92f35…"}
```

- **Members:** `type` (`about:blank` unless a `Problem` says otherwise), `title` (the RFC 9110
  status phrase), `status`, `detail` when there's something to tell, `instance` (the request's
  path, percent-encoded) and `request_id` when `RequestIDMiddleware` gave the request one.
- **HTTPExceptions:** FastAPI's, Starlette's and their subclasses (`BreakerOpen`, the rate
  limiter's 429…) keep their headers: `Retry-After`, `WWW-Authenticate`, `Allow`. A 4xx's detail
  is sent, since it's written for clients (`expose_detail=False` keeps it); a 5xx's never is: an
  `HTTPException(500, str(error))` becomes a bare 500, and the hidden detail is logged as a
  warning. A detail that isn't a string goes in an `errors` member. Codes under 400 (a redirect
  raised as an exception) keep FastAPI's answer.
- **Validation errors:** a 422 whose `errors` list each error's `loc`, `msg` and `type`, never
  its `input` (FastAPI echoes what was sent, passwords included) nor its `ctx`. At most
  `max_errors=100`, so that a huge invalid body doesn't get a huge answer.
- **Unhandled exceptions:** a 500 that says nothing about them. Starlette raises the exception
  again after the response, so the server logs it with its traceback (uvicorn's "Exception in
  ASGI application", with the request ID when you use `RequestIDLogFilter`) and `TestClient`
  raises it in tests; it isn't logged a second time. With `debug=True` Starlette shows its
  traceback page instead.
- **Problem:** an `HTTPException` with a `type` URI, a `title` and extension members, whose names
  are checked (three or more letters, digits or `_`, starting with a letter; never a standard
  member) and whose values are encoded when it's raised. Its detail is always sent, also with a
  5xx: you wrote it for the client. `problems.response(request, Problem(...))` answers one from
  your own exception handlers.
- **Alcazaba's middlewares too:** what they answer themselves (the 504 of a timeout, the 413 of a
  body too large, the 503 of maintenance, the 403 of CSRF, the idempotency errors) is a problem
  as well, headers such as `Retry-After` kept, once it's installed on the app.
- Handlers you register for a status code or an exception subclass still win; install it before
  the app starts.

<!-- --8<-- [end:problem-details] -->

## Request IDs
<!-- --8<-- [start:request-ids] -->

`RequestIDMiddleware` gives every request an ID, to find all the log lines of one request and to
let a client quote it when it reports an error:

```python
import logging

from fastapi_alcazaba.observability import RequestIDLogFilter, RequestIDMiddleware, request_id

app.add_middleware(RequestIDMiddleware)  # last: the outermost

handler = logging.StreamHandler()
handler.addFilter(RequestIDLogFilter())
handler.setFormatter(logging.Formatter("%(asctime)s [%(request_id)s] %(name)s: %(message)s"))
logging.getLogger().addHandler(handler)


@app.post("/jobs")
async def create_job() -> dict[str, str | None]:
    return {"ticket": request_id()}
```

With uvicorn, add the filter to its handlers too (`"filters": {"request_id": {"()":
"fastapi_alcazaba.observability.RequestIDLogFilter"}}` in its `log_config`), and its access and
error lines carry the ID.

- **Incoming IDs:** the request's `X-Request-ID` (`header=` for another) is reused when a proxy
  or the client set one that's safe to log: a single value matching `[A-Za-z0-9._:-]{1,128}`.
  Anything else (line breaks that would forge log lines, an endless value, several values) gets
  a new random one, 128 bits in hex. `trust_incoming=False` always makes a new one.
- **Everywhere in the request:** `request_id()` returns it in endpoints, dependencies, sync
  endpoints' threads and WebSockets; None outside a request. The response carries it in the
  same header (unless the app set that header itself), and so do WebSocket handshakes.
- **In errors:** after an unhandled exception the ID stays set for the rest of the request's
  task, so Starlette's 500 handler and the server's error log, which run outside every
  middleware, still know which request failed. `FastAPIProblemDetails` puts it in every problem,
  500s included, and in their header.
- **In logs:** `RequestIDLogFilter` sets `record.request_id` (`default="-"` outside a request; a
  `request_id` passed in `extra=` is kept). Add it to handlers: a logger's filters don't see the
  records of its child loggers.
- Add it last, the outermost, so that the answers of the other middlewares (a 413, a 504) carry
  the ID too.

<!-- --8<-- [end:request-ids] -->

## Audit trail
<!-- --8<-- [start:audit-trail] -->

`FastAPIAudit` records the sensitive actions: who did what to which resource, whether it worked,
and the request it came from. As a dependency, it writes the event when the endpoint ends:

```python
from typing import Annotated

from fastapi import Depends

from fastapi_alcazaba import FastAPIAudit
from fastapi_alcazaba.audit import AuditEntry

audit = FastAPIAudit(signing_key=settings.audit_signing_key)
deleting = audit.action("invoice.delete", resource="invoice:{invoice_id}")


@app.delete("/invoices/{invoice_id}", status_code=204)
async def delete_invoice(
    invoice_id: int,
    entry: Annotated[AuditEntry, Depends(deleting)],  # first: it records refusals too
    user: Annotated[User, Depends(current_user)],
) -> None:
    entry.actor = user.id
    entry.details["amount"] = await invoices.delete(invoice_id, owner=user.id)
```

- **How it ended:** `success` when the endpoint returns, `denied` when it raises a 401 or a 403
  (1008 for WebSockets), `failure` otherwise, with the exception's type and status, never its
  message: messages carry data. Declared before the dependency that authenticates, the audit
  dependency also records the requests that one refuses, with the resource from the path. An
  endpoint that answers an error without raising sets `entry.outcome`; `entry.reason` says why.
- **Outside endpoints:** `await audit.log("user.login", outcome="failure", reason="wrong
  password", request=request)` writes one event, and `async with audit.record("backup.restore",
  actor="cron")` one for a block of code (a job, a task), which fails when it raises or is
  cancelled.
- **Correlation:** each event has the request ID (with `RequestIDMiddleware`), the OpenTelemetry
  trace, the method, the route template (`/invoices/{invoice_id}`, never the path, which may
  carry tokens) and the client's address as the server reports it (see `client_ip` behind a
  proxy).
- **No secrets:** in `details`, the values under keys that name one (`password`, `newPassword`,
  `X-Api-Key`, `refresh_token`, `session_id`, `cookie`, `card_number`…: `SENSITIVE_WORDS`, plus
  yours with `redact=["iban"]`) and the strings that look like credentials (`Bearer …`, JWTs,
  private keys) are recorded as `[redacted]`. Objects and bytes are recorded by their type, not
  their content; strings are cut at 500 characters, lists and dicts at 50 items and 4 levels.
  Give `actor` and `resource` IDs, not e-mail addresses.
- **Where they go:** to the logger `fastapi_alcazaba.audit.events`, at INFO, one line of JSON
  per event, ASCII only so that nothing in it splits a line: give it a handler of its own and let
  the log pipeline ship it to storage that can't be rewritten. If that logger outputs nothing (no
  handler, the root logger at WARNING), it says so once. `sinks=` takes others: any object with
  `async def write(event)`, such as one inserting into a table, and `MemoryAuditSink` for tests.
- **Signed and chained:** with a `signing_key` (32 bytes or more, the same in every process; a
  list to rotate it, the first signs), each event is signed with HMAC-SHA256 and carries the
  signature of the previous one of its stream (one per process), numbered.
  `verify_audit_trail(events, key)` lists the events altered, forged, missing or out of their
  chain, from `AuditEvent`s or the JSON lines read back. The signature covers
  `fastapi_alcazaba/audit2`, a NUL byte and the event's
  [RFC 8785](https://www.rfc-editor.org/rfc/rfc8785) canonical JSON without its `signature`
  (`fastapi_alcazaba.core.canonical_json`): any language with a JCS library checks it the
  same way, a SIEM or an auditor's script included. For that, details keep to I-JSON: integers
  beyond 2⁵³ are recorded as text, lone surrogates as U+FFFD. It can't tell that the newest events were
  removed, and whoever has the key can rewrite the trail: keep it from the people whose actions
  are recorded.
- **Delivery:** a sink that fails or takes longer than `timeout` (5 s) is logged at ERROR with the
  event, so that it isn't lost, and counted; the other sinks still get it, and an event is written
  even when its request is cancelled meanwhile. The action isn't undone or refused for it: when an
  action mustn't happen unrecorded, write the record in the action's own database transaction.

```python
import json

from fastapi_alcazaba.audit import verify_audit_trail

with open("audit.log") as lines:
    events = [json.loads(line) for line in lines]
problems = verify_audit_trail(events, settings.audit_signing_key)
```

<!-- --8<-- [end:audit-trail] -->

## Request timeouts
<!-- --8<-- [start:request-timeouts] -->

`RequestTimeoutMiddleware` gives every request a deadline, so that a hung dependency or a client
that trickles its body doesn't hold a request (and its database connection, its memory…)
forever:

```python
from fastapi_alcazaba.observability import RequestTimeoutMiddleware

app.add_middleware(
    RequestTimeoutMiddleware,
    timeout=30,
    paths={"/reports": 120, "/events": None},  # Server-Sent Events: no deadline
)
```

- **Before the response starts:** at the deadline the app is cancelled, its `finally` blocks and
  the exit of its `yield` dependencies run, and the client gets a 504
  (`{"detail": "Request timed out"}`, or a problem with `FastAPIProblemDetails`). Reading the
  request body counts: a client that sends it a
  byte at a time runs out of time too.
- **While a response streams:** it's too late for a 504. The app is cancelled all the same and
  the TimeoutError raised again, so the server closes the connection: the client sees an
  incomplete body (an unfinished chunked read), never a short one that looks complete. Give
  streaming endpoints their own deadline in `paths`, or None.
- **After the response:** once its last byte is sent the deadline is off, so background tasks run
  as long as they need.
- **Sync endpoints** run in a thread that can't be interrupted: the client gets its 504 on time,
  but the thread runs to its end, after the exit of its `yield` dependencies has run. Endpoints
  that may hit the deadline are better async.
- **Paths:** `paths` sets other deadlines by path prefix (the longest wins, None for none),
  matched like the routes, without the `root_path`; `timeout=None` leaves the other paths
  without one. The app's own TimeoutErrors go through untouched, and WebSockets and lifespan
  events too.
- Both cases are logged as warnings with the method and the route template. A plain ASGI
  middleware: streaming responses still stream.

<!-- --8<-- [end:request-timeouts] -->

## Health checks
<!-- --8<-- [start:health-checks] -->

`health_router()` adds the two endpoints that load balancers and orchestrators probe:

```python
from fastapi_alcazaba.observability import HealthCheck, health_router


async def database() -> None:
    async with engine.connect() as connection:
        await connection.execute(text("SELECT 1"))


app.include_router(
    health_router({"database": database, "search": HealthCheck(ping_search, critical=False)})
)
```

Point the liveness probe at `/health/live` and the readiness probe (or the load balancer's
health check, e.g. the target group of an ECS service) at `/health/ready`.

- **Live vs ready:** `/health/live` only says that the process answers. `/health/ready` runs the
  checks concurrently, each with a timeout (`timeout=2.0`, or the check's own), and answers 200
  (`ok`, or `degraded` when only non-critical checks fail) or 503 (`fail`) with each check's
  status and duration. Keep dependencies out of liveness: a probe that checks the database gets
  healthy containers restarted every time it blips. A check passes when it returns anything but
  `False` without raising. Sync checks run in a daemon thread of their own, off the event loop,
  so one that hangs holds up neither the shutdown nor the threads of the loop's default executor;
  it isn't started again, under the same name in the same router, until its thread finishes.
- **The cache is checked, but isn't critical:** the cache's backend (`cache=`, else the app's or
  `FastAPICache`'s) is asked for a key under its prefix, straight, not through the circuit
  breaker. The cache fails open, so its outage makes the app `degraded`, not unready: taking
  every instance out of the load balancer would turn a Redis outage into a full one.
  `include_cache=False` leaves it out.
- **No leaked errors:** failures are logged with their traceback and never returned: a public
  probe mustn't tell the world the database's hostname or connection string.
- **Probes don't pile up:** concurrent `/ready` requests share one run of the checks, and its
  result answers the next ones for `reuse_for=1.0` seconds, so a flood of probes doesn't flood
  the database. A probe that goes away doesn't cancel the run the others are waiting for.
- The endpoints are left out of the OpenAPI schema unless `include_in_schema=True`; responses
  carry `Cache-Control: no-store`.

<!-- --8<-- [end:health-checks] -->

## Scheduled jobs
<!-- --8<-- [start:scheduled-jobs] -->

`FastAPIScheduler` runs jobs every N seconds or by a cron expression, in every process of the
app, and each tick once across them all:

```python
from fastapi_alcazaba import FastAPIScheduler

scheduler = FastAPIScheduler()  # FastAPICache's backend; or FastAPIScheduler(cache)


@scheduler.every(minutes=5)
async def refresh_catalog() -> None: ...


@scheduler.cron("0 3 * * *", tz="Europe/Madrid")
def nightly_report() -> None: ...  # sync jobs run in a thread


@scheduler.every(seconds=30, exclusive=False)
async def reload_settings() -> None: ...  # in every process: what each keeps in memory


app = FastAPI(lifespan=scheduler.lifespan)  # or `async with scheduler:` in your lifespan
```

- **Once per tick:** at each tick, the processes claim it in the backend, with an atomic counter
  named after the job and the tick, and only the first runs it. A lock held while the job runs
  keeps a run that outlasts its interval from overlapping the next tick in another process
  (memcached can't hold locks: there, each tick still runs once, but long runs may overlap).
  The ticks are the same in every process: `every(minutes=5)` runs at :00, :05… (UTC), so keep
  the servers' clocks in sync (NTP). `exclusive=False` runs a job in every process.
- **Cron:** five fields (minute, hour, day of the month, month, day of the week, Sunday being 0 or
  7) with `*`, lists, ranges, steps and names (`*/15 9-17 * * mon-fri`), or `@hourly`, `@daily`,
  `@weekly`, `@monthly` and `@yearly`, in the time zone `tz` (UTC by default). When the clocks
  change it does what cronie does: a job at a fixed time runs an hour later on the day its hour
  is skipped and once on the day it repeats; a job with `*` in its minute or hour follows real
  time. No dependency: the parser is Alcazaba's.
- **No drift, no bursts:** the next run is the next tick after the clock, not a delay after the
  last run ended; ticks missed while a run took longer than its interval are skipped, not
  run in a row afterwards.
- **Failures:** an exception is logged with its traceback, counted, and the job keeps its
  schedule. If the backend fails, the tick is skipped (`fail_open=True` runs it anyway, maybe in
  every process).
- **Start and stop:** the jobs start with the app and never hold up its startup. At shutdown,
  jobs waiting for their next run stop at once and running ones get `grace` seconds (10) before
  they're cancelled; a sync job's thread runs to its end. `scheduler.next_run(name)` says when a
  job runs next. A job is named after its module and function (`name=` for another), the name
  that its ticks have in the backend.

<!-- --8<-- [end:scheduled-jobs] -->

## Task queue
<!-- --8<-- [start:task-queue] -->

`FastAPITaskQueue` hands work over from an endpoint to run in the background, in this process or
in any other that shares the cache's Redis or Valkey:

```python
from fastapi_alcazaba import FastAPITaskQueue

queue = FastAPITaskQueue()  # FastAPICache's backend; or FastAPITaskQueue(cache)


@queue.task(retries=5)
async def send_welcome_email(user_id: int) -> None: ...


app = FastAPI(lifespan=queue.lifespan)  # this process runs tasks too


@app.post("/users")
async def create_user(user: NewUser) -> None:
    ...
    await queue.enqueue(send_welcome_email, user.id)  # delay=60 or a timedelta to wait first
```

- **At least once:** tasks live in a Redis stream read by a consumer group, and leave it only once
  they've finished. A process that dies while running one, or can't reach the backend to say it
  finished, means that it runs again elsewhere: write tasks so that running one twice does no harm
  (check first whether the e-mail went out). A worker that stops answering has its tasks taken
  over after `visibility_timeout` (60 s); running tasks are kept from that by heartbeats, however
  long they run. A task handed out more than `max_deliveries` times (5) goes to the dead letters
  instead: it may be what kills the workers.
- **Only registered tasks run, with JSON arguments**, never pickle. `enqueue` checks that the
  arguments fit the function and serialise (TypeError otherwise), up to `max_payload_size` bytes
  (64 KiB); they arrive as JSON gives them back (tuples as lists).
- **Sealed:** with a `signing_key` or `encryption_key` on the cache, every task is signed or
  encrypted and bound to its queue. A task that someone planted in the backend or altered is
  refused, logged and sent to the dead letters, never run. Without a key a warning is logged at
  start: whoever can write to the backend can run your tasks with arguments of their choosing.
- **Retries:** a task that raises is tried again `retries` times (3), `backoff` seconds after the
  first failure (1), twice that after the second… up to `max_backoff` (600), each with jitter so
  that tasks that failed together (a service down) don't all retry at once. Then it goes to the
  dead letters with the type of its error, never the message, which may hold personal data.
  `dead_letters()` lists them, `requeue_dead_letters()` puts them back with their attempts reset,
  and `discard_dead_letters(ids)` deletes them.
- **Delayed tasks and retries** wait in a sorted set, and move to the stream when they're due in
  a Lua script: two processes never both move one.
- **Bounded:** the queue holds at most `max_length` tasks (100,000: waiting, running or delayed),
  and `enqueue` raises `QueueFull` beyond, instead of dropping any; about `max_dead_letters`
  (10,000) are kept. `pending()` counts the tasks that haven't finished.
- **Workers:** `start()` and `stop()`, `lifespan` or `async with`, running `concurrency` tasks at a
  time (10) in each process; sync functions run in a thread. `stop()` takes no more tasks and gives
  the running ones `grace` seconds (10); those still running then are cancelled and go back to the
  queue at once (a sync task's thread can't be stopped: it runs to its end, so that task may run
  twice). A process that only runs tasks: `async with queue: await asyncio.Event().wait()`.
- **Backends:** Redis 6.2+ or Valkey through `RedisBackend` or `ValkeyBackend` (or a
  `TieredBackend` in front of one), clusters included: a queue's keys share a hash tag. Other
  backends raise `NotImplementedError`. Its keys are state keys, which `clear()` never deletes.
- **Small on purpose:** no results, priorities or workflows. Each run is the root of a trace of its
  own, linked to the span of the request that enqueued it.

<!-- --8<-- [end:task-queue] -->

## Circuit breakers
<!-- --8<-- [start:circuit-breakers] -->

`FastAPIBreaker` stops calling a dependency that keeps failing (a database, an SDK, a service
reached without httpx), so that requests fail at once instead of waiting for it, and it gets room
to recover:

```python
from fastapi_alcazaba import FastAPIBreaker

payments = FastAPIBreaker("payments", failure_threshold=5, reset_timeout=30, timeout=10)


@payments
async def charge(order: Order) -> Receipt:
    return await payments_sdk.charge(order.id, order.total)


async def refund(order: Order) -> None:
    async with payments.attempt():  # or around a block
        await payments_sdk.refund(order.id)
```

- **States:** after `failure_threshold` consecutive failures it opens, and for `reset_timeout`
  seconds calls fail with `BreakerOpen` without reaching the dependency. Then a single trial call
  goes through: its success closes the breaker, its failure opens it again.
- **In FastAPI:** `BreakerOpen` is an `HTTPException`: an endpoint that hits an open breaker
  answers 503 with `Retry-After` (the seconds until the trial), without naming the dependency.
- **What counts:** exceptions in `failures` (any `Exception` by default) are failures; those in
  `exclude` mean the dependency answered (a "not found" from its SDK) and are successes; others,
  and a cancelled call, give no verdict. `timeout` bounds async calls: a slower call is
  cancelled, raises `TimeoutError` and counts as a failure. Sync code can't be interrupted, so a
  breaker with a timeout refuses to guard it.
- **Per process:** each process keeps its own state, with no round trip on every call. Sync
  functions work too, also from the threads of FastAPI's pool.
- **Closed by hand:** `payments.reset()` closes it at once and forgets the failures, when you know
  the dependency is back (from an admin endpoint) or between tests; `failure_threshold` and
  `reset_timeout` can be read back.
- **Observability:** opening is logged as a warning and closing again as info, and both are
  counted with the calls (`fastapi_alcazaba.breaker.*` metrics). `breaker.state` says
  `"closed"`, `"open"` or `"half-open"`, handy in a health check:
  `HealthCheck(lambda: payments.state != "open", critical=False)`.

<!-- --8<-- [end:circuit-breakers] -->

## Concurrency limits
<!-- --8<-- [start:concurrency-limits] -->

`FastAPIConcurrencyLimiter` limits how many requests run *at the same time*, across every process
and container, for endpoints whose cost is how long they run (AI, OCR, PDF rendering). The rate
limiter counts requests per window; this counts the ones in flight:

```python
from fastapi_alcazaba.rate_limit import FastAPIConcurrencyLimiter
from fastapi_alcazaba.rate_limit import client_ip

ocr = FastAPIConcurrencyLimiter(4, name="ocr", wait=2)  # FastAPICache's backend, or cache=...


@app.post("/invoices/scan", dependencies=[Depends(ocr)])
async def scan(file: UploadFile) -> Invoice: ...


@app.post("/summaries", dependencies=[Depends(FastAPIConcurrencyLimiter(1, identifier=client_ip))])
async def summarise(text: str) -> Summary: ...  # one at a time per client


async def scan_pending() -> None:  # outside requests: a job, a queue consumer
    async with ocr.slot():
        ...
```

- **Slots in the backend:** `limit` locks (up to 100), tried in random order so that the
  processes don't all queue on the first; a request holds one while it runs, renewed every
  `ttl / 3` seconds, and the slots of a process that dies expire after `ttl` (60). Keys are
  SHA-256 digests: client identities never reach the backend in clear.
- **When it's full:** a request waits up to `wait` seconds (0: tries once), polling with
  jittered backoff, then gets a 503 with `Retry-After` (`retry_after`, 1 s by default); a 429 when
  the limit is per client (`identifier`). There's no queue and no order among the waiters.
  WebSockets are closed with 1013 (1008 per client) and hold their slot while connected.
- **Sharing:** by default each route (methods and path template) has its own slots; limiters with
  the same `name` and `limit` share them. `slot(key=...)` gives each key its own slots outside
  requests (it needs a `name`), and `slot(connection=request)` takes one around just the
  expensive part of an endpoint.
- **Streaming:** the slot is released when FastAPI runs the dependency's exit code. Since FastAPI
  0.118 that's after the response is sent, so a `StreamingResponse` keeps its slot until its last
  byte; `Depends(limiter, scope="function")` (FastAPI 0.121) releases it when the endpoint
  returns, before the body. FastAPI 0.115 to 0.117 release it before the response is sent: take
  a `slot()` inside the stream there.
- **Failures:** if the backend fails, each process counts the slots on its own (`fail_open=True`,
  the default: at most `limit` per process run); `fail_open=False` refuses with a 503 instead
  (`slot()` raises `BackendUnavailable`). memcached can't hold locks.
- **Limits:** like any lock on a cache-like store, a holder that stalls for longer than `ttl`
  loses its slot while it still runs, and someone else may take it. Taking a slot when most are
  busy costs a round trip per busy slot tried.

<!-- --8<-- [end:concurrency-limits] -->

## Leader election
<!-- --8<-- [start:leader-election] -->

`FastAPILeader` elects one process among all those sharing the backend, for work that must run in
exactly one of them all the time (a queue consumer, an outbox relay, a watcher) rather than at
given times (that's the scheduler):

```python
from fastapi_alcazaba.background import FastAPILeader

leader = FastAPILeader("outbox")  # FastAPICache's backend; or FastAPILeader("outbox", cache=cache)


@leader.run
async def relay_outbox() -> None:
    while True:
        await relay_pending()
        await asyncio.sleep(1)


app = FastAPI(lifespan=leader.lifespan)  # or `async with leader:` in your lifespan

...
if leader.is_leader:  # anywhere, sync code included
    ...
```

- **Election:** every process campaigns for a lock named after the leader, renewed every
  `ttl / 3` seconds (`ttl` is 15 by default). The one that takes it leads and runs the functions
  given to `run`; the others try again every `ttl / 3` seconds or so, with jitter.
- **Hand-over:** a leader that stops releases the lock, and another process takes over within
  `ttl / 3` seconds; one that dies, within `ttl` and a third. A leader that loses the lock (its
  renewals couldn't keep up) cancels its functions within a second and campaigns again; at the
  next election they run again.
- **Functions:** async, without arguments, added before `start()`. They're cancelled when the
  leadership ends and must let the cancellation through. An exception is logged with its
  traceback; the process keeps leading.
- **Failures:** while the backend is down nobody is elected, and a leader that can't renew steps
  down when its lock expires: no leader is better than two. memcached can't hold locks.
- **Limits:** a leader that stalls for longer than `ttl` (a long pause, a network partition) can
  overlap the next one for a moment. Check `is_leader` between the steps of long work, and keep
  that work idempotent.

<!-- --8<-- [end:leader-election] -->

## Maintenance mode
<!-- --8<-- [start:maintenance-mode] -->

`FastAPIMaintenance` is a switch shared by every process; while it's on, `MaintenanceMiddleware`
answers 503 with `Retry-After` to every request but the allowed ones:

```python
from fastapi_alcazaba.rate_limit import client_ip
from fastapi_alcazaba.operations import FastAPIMaintenance, MaintenanceMiddleware

maintenance = FastAPIMaintenance()  # FastAPICache's backend; or FastAPIMaintenance(cache)
app.add_middleware(
    MaintenanceMiddleware,
    maintenance=maintenance,
    allow_paths=("/health", "/admin"),  # always served
    allow=lambda request: client_ip(request) in OFFICE,  # the team keeps working
)

await maintenance.enable("Back at 10:30", retry_after=900)  # from an admin route, a script...
await (
    maintenance.status()
)  # MaintenanceStatus(enabled=True, message=..., retry_after=900, since=...)
await maintenance.disable()
```

- **Shared, but not asked on every request:** the state lives in the backend, and each process
  keeps a copy refreshed at most every `refresh` seconds (1 by default), so a change reaches every
  process within about that. `status()` reads the backend now; `forget()` drops the copy, and a
  new cache drops it by itself (each test's lifespan sets one up).
- **Responses:** a 503 whose `detail` is the message ("Service Unavailable" without one), with
  `Retry-After` when given and `Cache-Control: no-store`. WebSockets are closed with 1013 (try
  again later). Lifespan events go through. A pure ASGI middleware: streaming isn't affected.
- **Exceptions:** `allow_paths` are prefixes (`/health` covers `/health/live`, not `/healthz`),
  matched without the `root_path` of a proxy or a mount, and are served without reading the state.
  `allow` (sync or async) receives the request: use the address the server reports
  (`client_ip`), never `X-Forwarded-For`.
- **Failures:** when the backend fails, each process keeps the last state it read (maintenance
  often means work on the backend itself: its failure mustn't reopen an app closed on purpose)
  and serves normally if it never read one. `enable()` and `disable()` raise
  `BackendUnavailable`: nothing changes then.
- **Integrity:** with a `signing_key` or `encryption_key` on the cache the state is sealed, and a
  state planted in the backend by someone else is ignored; the message is capped at 1000
  characters.

<!-- --8<-- [end:maintenance-mode] -->

## Feature flags
<!-- --8<-- [start:feature-flags] -->

`FastAPIFeatureFlags` keeps flags in the backend: turn features on and off for everybody, for some
users or tenants, or for a percentage of them, from any process and without a deploy:

```python
from fastapi_alcazaba.operations import FastAPIFeatureFlags

flags = FastAPIFeatureFlags()  # FastAPICache's backend; or FastAPIFeatureFlags(cache, refresh=5)

await flags.set("new-checkout", enabled=True, percentage=10, allow=["user-42"], deny=["user-7"])

if await flags.enabled("new-checkout", subject=str(user.id)):
    ...


def user_id(request: Request) -> str | None:
    return request.headers.get("x-user-id")


@app.get("/beta/report", dependencies=[Depends(flags.require("beta-report", subject=user_id))])
async def beta_report() -> Report: ...  # 404 while the flag is off for the user
```

- **Rules:** a flag is on for a subject when it's enabled, the subject isn't in `deny`, and it's
  in `allow` or falls within `percentage`. No percentage means everybody, but only `allow` when
  there's an allow list: a flag for a few never goes to everyone by mistake. `enabled=False` turns it off for
  everyone, `allow` included. Without a subject, a partial rollout counts as off. An undefined
  flag gives `default` (False).
- **Sticky rollouts:** a subject's place comes from a hash of the flag's name and the subject, so
  it keeps its answer while the percentage stays, raising the percentage only adds subjects, and
  each flag picks different subjects.
- **Hidden when off:** `require()` answers FastAPI's own `404 Not Found`, so the flag isn't
  revealed, and closes WebSockets with 1008. For subjects that need other dependencies (the
  current user), call `enabled()` from your own dependency.
- **Local copies:** each process keeps the definitions it reads for `refresh` seconds (5), so a
  change reaches every process within that, and a flag costs at most one round trip per
  `refresh`. When the backend fails, the last definition read is used, or the default.
  `get(name)` reads the backend now; `delete(name)` removes a flag; `forget()` drops the copies,
  and a new cache drops them by itself.
- **Privacy and integrity:** subjects are stored as SHA-256 digests salted with the flag's name,
  never in clear (at most 10,000 per list). Whoever reads the backend can still confirm a guess
  of a guessable subject (an e-mail address): prefer opaque ids, or give the cache an
  `encryption_key`, which encrypts the definitions. With a `signing_key` or `encryption_key`,
  definitions planted by someone else are ignored. Names are up to 100 letters, digits, `.`, `_`
  and `-`.

<!-- --8<-- [end:feature-flags] -->

## Resilience
<!-- --8<-- [start:resilience] -->

The cache never makes a request fail. Every backend call has a timeout (`backend_timeout`, 1 s by
default): if it runs out, or the backend fails, the request goes on without cache. After
`circuit_breaker_failures` consecutive failures the backend isn't called at all for
`circuit_breaker_reset` seconds, so that requests don't each wait for the timeout; then one call
tries again.

Concurrent misses for the same key compute once per process (`single_flight`). With several
processes or containers, `distributed_lock=True` makes the first one take a lock in Redis/Valkey
while the others wait for its value (up to `lock_wait` seconds). And `early_refresh=0.1` lets
callers refresh an entry, with growing probability, during the last 10 % of its lifetime: popular
entries are renewed before they expire, instead of every caller missing at once. If that refresh
fails, the cached value is served.

<!-- --8<-- [end:resilience] -->

## Metrics and traces
<!-- --8<-- [start:metrics-and-traces] -->

Alcazaba logs under `fastapi_alcazaba.*` with a `NullHandler`, as libraries do: its warnings (a
breaker that opens, maintenance turned on, a backend that fails) show once the app configures
logging, e.g. `logging.basicConfig(level=logging.INFO)`. uvicorn alone only configures its own
loggers.

Alcazaba only uses the OpenTelemetry API (the `otel` extra): it records nothing until the app
configures an SDK, and never configures one itself. The simplest way is OpenTelemetry's own
zero-code instrumentation, with the standard variables:

```bash
pip install opentelemetry-distro opentelemetry-exporter-otlp
opentelemetry-bootstrap -a install  # instrumentations for FastAPI, httpx2, redis, SQLAlchemy…

OTEL_SERVICE_NAME=shop-api \
OTEL_RESOURCE_ATTRIBUTES=deployment.environment.name=production \
OTEL_EXPORTER_OTLP_ENDPOINT=https://collector.internal:4317 \
OTEL_PYTHON_LOG_CORRELATION=true \
OTEL_PYTHON_FASTAPI_EXCLUDED_URLS="health/live,health/ready" \
opentelemetry-instrument uvicorn app:app
```

Give the collector's endpoint explicitly, over TLS: traces and logs carry what your app does.
Then Alcazaba adds to them:

- **Spans:** one per cached call (`fastapi_alcazaba <function>`, with `fastapi_alcazaba.result`:
  hit, miss, refresh or bypass) and, inside it, one per backend call (`fastapi_alcazaba.backend
  get`, `set`, `lock`…, with failures recorded); instrumented Redis/Valkey clients nest their own
  inside. Each run of a scheduled job is the root of a trace of its own
  (`fastapi_alcazaba.scheduler <job>`, failures recorded), so what the job does (queries, calls)
  is grouped under it instead of scattered. So is each run of a task (`fastapi_alcazaba.task
  <task>`, with its queue, ID and attempt), linked to the span that enqueued it.
- **On the request's span:** the events `fastapi_alcazaba.rate_limit` (a request limited or
  unchecked), `fastapi_alcazaba.breaker.refused`, `.opened` and `.closed`,
  `fastapi_alcazaba.outbound.refused` (an outbound call to an origin whose breaker is open) and
  `fastapi_alcazaba.csrf.refused`, and the attribute `fastapi_alcazaba.idempotency` with the
  outcome of an Idempotency-Key.
- **Never:** keys, tokens, identities, codes or request bodies. Attributes are names, routes and
  outcomes, few enough to keep the cardinality of the metrics low.

| Metric | Attributes |
| --- | --- |
| `fastapi_alcazaba.lookups` | `fastapi_alcazaba.result` (hit, miss, refresh, bypass), `fastapi_alcazaba.namespace` |
| `fastapi_alcazaba.stores` | `fastapi_alcazaba.outcome` (stored, uncacheable, too_large, unencodable, failed), `fastapi_alcazaba.namespace` |
| `fastapi_alcazaba.entry.size` | `fastapi_alcazaba.namespace` |
| `fastapi_alcazaba.backend.duration` | `fastapi_alcazaba.operation` (get, set, delete, incr, lock, unlock, extend_lock, tag, incr_window) |
| `fastapi_alcazaba.backend.errors` | `fastapi_alcazaba.operation`, `error.type` |
| `fastapi_alcazaba.rate_limit` | `fastapi_alcazaba.outcome` (allowed, limited, unchecked), `fastapi_alcazaba.limiter` (its name, or the route) |
| `fastapi_alcazaba.breaker.calls` | `fastapi_alcazaba.outcome` (success, failure, refused), `fastapi_alcazaba.breaker` |
| `fastapi_alcazaba.breaker.transitions` | `fastapi_alcazaba.state` (open, closed), `fastapi_alcazaba.breaker` |
| `fastapi_alcazaba.scheduler.runs` | `fastapi_alcazaba.outcome` (ran, failed, skipped, overlapping, unchecked), `fastapi_alcazaba.job` |
| `fastapi_alcazaba.idempotency` | `fastapi_alcazaba.outcome` (stored, replayed, conflict, mismatch, not_kept, invalid, too_large, unchecked) |
| `fastapi_alcazaba.tokens.checks` | `fastapi_alcazaba.outcome` (valid, revoked, unchecked) |
| `fastapi_alcazaba.otp.verifications` | `fastapi_alcazaba.outcome` (success, wrong, missing, exhausted, malformed, unchecked): a burst of wrong or exhausted is someone guessing |
| `fastapi_alcazaba.lock.wait` | `fastapi_alcazaba.acquired` (true, false): seconds spent acquiring locks |
| `fastapi_alcazaba.lock.lost` | locks whose renewals couldn't keep up |
| `fastapi_alcazaba.health.checks` | `fastapi_alcazaba.check`, `fastapi_alcazaba.status` (ok, fail) |
| `fastapi_alcazaba.health.check.duration` | `fastapi_alcazaba.check` |
| `fastapi_alcazaba.concurrency` | `fastapi_alcazaba.outcome` (acquired, local: counted in the process while the backend failed, refused, unavailable), `fastapi_alcazaba.limiter` |
| `fastapi_alcazaba.leader.transitions` | `fastapi_alcazaba.leader`, `fastapi_alcazaba.event` (elected, lost, resigned) |
| `fastapi_alcazaba.maintenance.refused` | requests refused during maintenance |
| `fastapi_alcazaba.flags.evaluations` | `fastapi_alcazaba.flag`, `fastapi_alcazaba.result` (on, off, default) |
| `fastapi_alcazaba.problems` | `fastapi_alcazaba.status`, `fastapi_alcazaba.kind` (http_exception, problem, validation, unhandled) |
| `fastapi_alcazaba.request_timeouts` | `fastapi_alcazaba.outcome` (504, cut: already streaming), `fastapi_alcazaba.route` |
| `fastapi_alcazaba.request_ids` | `fastapi_alcazaba.source` (incoming, generated, rejected): many rejected may be a misconfigured proxy, or someone forging log lines |
| `fastapi_alcazaba.outbound.breaker` | `fastapi_alcazaba.event` (opened, refused), `server.address` |
| `fastapi_alcazaba.outbound.ssrf.blocked` | `fastapi_alcazaba.reason` (scheme, host, port, address) |
| `fastapi_alcazaba.webhooks` | `fastapi_alcazaba.outcome` (accepted, replayed, rejected, malformed, too_large, unchecked, unavailable) |
| `fastapi_alcazaba.api_keys` | `fastapi_alcazaba.outcome` (accepted, malformed, unknown, wrong_secret, expired, insufficient_scope) |
| `fastapi_alcazaba.auth` | `fastapi_alcazaba.strategy`, `fastapi_alcazaba.outcome` (success, failure: credentials there but refused, unauthenticated: no strategy found a user) |
| `fastapi_alcazaba.authz` | `fastapi_alcazaba.outcome` (allowed, denied, unauthenticated): a burst of denied is someone probing |
| `fastapi_alcazaba.csrf.refused` | `fastapi_alcazaba.kind` (http, websocket), `fastapi_alcazaba.reason` (fetch_site, origin, no_source, token) |
| `fastapi_alcazaba.sessions` | `fastapi_alcazaba.event` (created, updated, rotated, deleted, touched, rejected: forged or altered, unreadable, unavailable) |
| `fastapi_alcazaba.uploads` | `fastapi_alcazaba.outcome` (accepted, too_large, too_many_pixels, unsupported, unreadable, empty) |
| `fastapi_alcazaba.uploads.size` | `fastapi_alcazaba.content_type` (the detected one): bytes of the uploads accepted |
| `fastapi_alcazaba.tasks.enqueued` | `fastapi_alcazaba.queue`, `fastapi_alcazaba.task` |
| `fastapi_alcazaba.tasks` | `fastapi_alcazaba.queue`, `fastapi_alcazaba.task`, `fastapi_alcazaba.outcome` (succeeded, retried, failed, interrupted; for refused tasks malformed, tampered, too_large, unknown_task, invalid_arguments, abandoned, with the task only when it's registered) |
| `fastapi_alcazaba.tasks.duration` | `fastapi_alcazaba.queue`, `fastapi_alcazaba.task`, `fastapi_alcazaba.outcome` |
| `fastapi_alcazaba.tasks.reclaimed` | `fastapi_alcazaba.queue`: tasks taken over from workers that stopped answering |
| `fastapi_alcazaba.audit` | `fastapi_alcazaba.action`, `fastapi_alcazaba.outcome` (success, failure, denied): a burst of denied is someone probing |
| `fastapi_alcazaba.audit.undelivered` | `fastapi_alcazaba.sink`, `fastapi_alcazaba.reason` (error, timeout): events a sink didn't write |

<!-- --8<-- [end:metrics-and-traces] -->

## Responses in several languages
<!-- --8<-- [start:responses-in-several-languages] -->

If an endpoint translates its response according to a request header, such as
`Accept-Language`, add that header to `vary_headers` (`@cache(vary_headers=["accept-language"])`),
or a user would get the cached response in another user's language.

<!-- --8<-- [end:responses-in-several-languages] -->

## Pagination
<!-- --8<-- [start:pagination] -->

Twelve strategies, with nothing but FastAPI and pydantic. Each is configured once and used as a
dependency, which reads and checks its query parameters; the endpoint gets them as an argument, so
`@cache` keeps each page under a key of its own:

| Strategy | Query parameters | For |
| --- | --- | --- |
| `PaginationPageNumber` | `page`, `page_size` | Numbered pages, with the total |
| `PaginationLimitOffset` | `limit`, `offset` | Slices, with the total |
| `PaginationOffset` | `offset`, `limit` | Slices, without counting the whole list |
| `PaginationRange` | `start`, `end` | Ranges of items, with `Content-Range` |
| `PaginationCursor` | `cursor`, `limit` | An opaque, signed cursor: infinite scroll, feeds |
| `PaginationKeyset` | `after`, `before`, `limit` | Keyset pagination, both ways |
| `PaginationIdBased` | `last_id`, `limit` | The id of the last item seen |
| `PaginationTimeBased` | `timestamp`, `limit` | Before or after a time: feeds, syncing |
| `PaginationSearchAfter` | `search_after`, `limit` | The sort values of the last item, readable |
| `PaginationToken` | `page_token`, `page_size` | An opaque token, bound to the query, that expires |
| `PaginationScroll` | `scroll_id`, `batch_size` | A snapshot kept in the backend, read in batches |
| `PaginationStreaming` | none | Every item as it comes, as NDJSON |

```python
from fastapi_alcazaba.pagination.page_number import (
    PageNumberPage,
    PageNumberParams,
    PaginationPageNumber,
)

pages = PaginationPageNumber(default_page_size=20, max_page_size=100)


@app.get("/products")
@cache(expire=60, tags=["products"])
async def products(
    params: Annotated[PageNumberParams, Depends(pages)],
) -> PageNumberPage[Product]:
    return pages.paginate(await load_products(), params)
```

- **By position or by key:** page numbers, offsets and ranges count items, so a row added or
  deleted meanwhile shifts the pages (an item skipped or seen twice), and a database walks past
  every row before the page. Cursors, keysets, ids, timestamps and sort values point at the last
  item instead: the pages don't shift, and an index finds any of them as fast as the first.
- **Links:** the response links to the pages around it in the `Link` header
  ([RFC 8288](https://www.rfc-editor.org/rfc/rfc8288)), as GitHub's API does:
  `</products?page=3&page_size=20>; rel="next"`. The links keep the request's other query
  parameters and leave out the host, so they don't depend on the `Host` header nor on a proxy in
  front. Pages served from the cache get them too, made for the request they answer.
- **From a database:** `paginate` takes the whole list and finds the page in it, or the rows the
  database returned for the page: with their `total` for the strategies that count, or
  `queried=True` for the others, which fetch one row more (`params.fetch`) to tell whether
  there's a next page.
- **Limits:** each strategy has its own maximum (100 items by default), so that nobody asks for a
  million rows at once: anything beyond, or malformed, is a 422 before the endpoint runs. They're
  in the OpenAPI schema too.
- **Cached:** every page, size and filter is one entry: tag them (`tags=["products"]`) to drop
  them all when the data changes. For short lists, caching the whole list with `get_or_set` and
  slicing every page from it keeps a single entry. The
  [pagination example](https://github.com/nonodev96/fastapi-alcazaba/blob/main/examples/pagination/main.py)
  does it three ways, with the database session left out of the key (`exclude_params`).

<!-- --8<-- [end:pagination] -->

### PaginationPageNumber
<!-- --8<-- [start:pagination-page-number] -->

Numbered pages, with the total: `GET /products?page=2&page_size=20` answers
`{"items": [...], "total": 245, "page": 2, "page_size": 20, "pages": 13}`. For lists that people
browse page by page, jump through, and want to know the size of.

- **From a database:** query only the page, with `params.offset` and `params.limit`, and count:
  `pages.paginate(rows, params, total=count)`. Without `total`, `paginate` slices the page from
  the whole list.
- **Links:** first, previous, next and last.
- **Its cost:** counting the whole list on every page, and the shifting of pages by position.

<!-- --8<-- [end:pagination-page-number] -->

### PaginationLimitOffset
<!-- --8<-- [start:pagination-limit-offset] -->

Slices, with the total: `GET /products?limit=20&offset=40` answers
`{"items": [...], "total": 245, "limit": 20, "offset": 40}`. The same as page numbers for clients
that count in items rather than pages, and can ask for any slice.

```python
from fastapi_alcazaba.pagination.limit_offset import (
    LimitOffsetPage,
    LimitOffsetParams,
    PaginationLimitOffset,
)

slices = PaginationLimitOffset(default_limit=20, max_limit=100)


@app.get("/products")
async def products(
    params: Annotated[LimitOffsetParams, Depends(slices)],
) -> LimitOffsetPage[Product]:
    rows = await session.scalars(select(Product).limit(params.limit).offset(params.offset))
    return slices.paginate(rows.all(), params, total=await count_products())
```

- **Links:** first, previous, next and last, on the grid of `limit` from 0, where clients that
  start at the first page walk.
- **Without the total:** `PaginationOffset` skips the count.

<!-- --8<-- [end:pagination-limit-offset] -->

### PaginationOffset
<!-- --8<-- [start:pagination-offset] -->

Slices without the total: `GET /logs?offset=40&limit=20` answers `{"items": [...], "offset": 40,
"limit": 20, "has_more": true}`. The database is asked for one row more than the page
(`params.fetch`), which isn't returned but tells whether there's a next one, so the list is never
counted. For lists whose count is slow (big tables, searches, joins) or outdated as soon as it's
made, and for clients that only ever go to the next page:

```python
from fastapi_alcazaba.pagination.offset import OffsetPage, OffsetParams, PaginationOffset

offsets = PaginationOffset(default_limit=20, max_limit=100)


@app.get("/logs")
@cache(expire=10)
async def logs(params: Annotated[OffsetParams, Depends(offsets)]) -> OffsetPage[Line]:
    query = select(Line).order_by(Line.id.desc()).limit(params.fetch).offset(params.offset)
    rows = (await session.scalars(query)).all()
    return offsets.paginate(rows, params, queried=True)
```

- **Queried or sliced:** with `queried=True`, the rows are those of
  `LIMIT params.fetch OFFSET params.offset`, one more than `limit` when there's a next page.
  Without it, `paginate` slices the page from the whole list.
- **Links:** first, previous and, while `has_more`, next. There's no last page, nor a number of
  pages: for those, `PaginationLimitOffset` counts.
- **By position:** a row added or deleted before the page shifts it, so an item may be skipped
  or seen twice, and a database still walks past every row before the offset: deep pages get
  slower. For lists that grow while they're read, page by key (`PaginationIdBased`,
  `PaginationCursor`).
- **Limits:** `limit` up to `max_limit` and `offset` from 0, a 422 otherwise.

<!-- --8<-- [end:pagination-offset] -->

### PaginationRange
<!-- --8<-- [start:pagination-range] -->

Ranges of items by the indexes of the first and the last, from 0 and both included:
`GET /contacts?start=0&end=24` answers `{"items": [...], "start": 0, "end": 24, "total": 319,
"limit": 25}` with `Content-Range: items 0-24/319`. For data grids and tables that ask for the
rows they show, and read from the header which ones they got and how many there are:

```python
from fastapi_alcazaba.pagination.ranges import PaginationRange, RangePage, RangeParams

ranges = PaginationRange(default_items=25, max_items=100)


@app.get("/contacts")
@cache(expire=60, tags=["contacts"])
async def contacts(params: Annotated[RangeParams, Depends(ranges)]) -> RangePage[Contact]:
    query = select(Contact).order_by(Contact.name).limit(params.limit).offset(params.offset)
    rows = (await session.scalars(query)).all()
    return ranges.paginate(rows, params, total=await count_contacts())
```

- **The range returned:** `end` is the index of the last item in the response, which is short of
  the one asked for at the end of the list; `null`, and `Content-Range: items */319`, when there
  are none. `limit` is the size asked for, which the links keep.
- **From a database:** query the range with `params.offset` and `params.limit`, and count:
  `ranges.paginate(rows, params, total=count)`. Without `total`, `paginate` slices the range from
  the whole list.
- **Errors:** without `end`, a range holds `default_items`. One whose `end` is before its `start`,
  or that holds more than `max_items` items, is a 400 before the endpoint runs; one that starts
  past the last item of a list that isn't empty, a 416 with `Content-Range: items */<total>`.
- **Links:** first, previous, next and last ranges of the same size, the first and last on the
  grid of that size from 0. Pages served from the cache get their `Content-Range` too.
- **Its cost:** like page numbers, a count on every range, and pages that shift when rows are
  added or deleted before them.

<!-- --8<-- [end:pagination-range] -->

### PaginationCursor
<!-- --8<-- [start:pagination-cursor] -->

A cursor points at the last item of a page by its sort key, not by its position: rows added or
deleted meanwhile don't shift the pages (no item skipped or seen twice), and a database finds any
page through an index, as fast as the first. For infinite scroll, feeds and clients that sync
everything:

```python
from fastapi_alcazaba.pagination.cursor import CursorPage, CursorParams, PaginationCursor

cursors = PaginationCursor(secret=settings.cursor_secret)  # secrets.token_urlsafe(32)


@app.get("/events")
@cache(expire=60)
async def events(params: Annotated[CursorParams, Depends(cursors)]) -> CursorPage[Event]:
    query = select(Event).order_by(Event.created_at, Event.id).limit(params.fetch)
    if params.after is not None:  # (created_at, id) of the last event seen
        query = query.where(tuple_(Event.created_at, Event.id) > params.after)
    rows = (await session.scalars(query)).all()
    return cursors.paginate(rows, params, key=lambda e: (e.created_at, e.id), queried=True)
```

`GET /events?limit=20` answers `{"items": [...], "limit": 20, "next_cursor": "WyIyMDI2..."}`,
and `?cursor=<next_cursor>` the next page; `next_cursor` is null on the last one. The `Link`
header has the first page and, except on the last page, the next one.

- **The sort key** (`key`) must be unique: `(created_at, id)`, not `created_at` alone, which two
  items may share. Its values can be strings, integers (up to 64 bits), floats, booleans,
  datetimes, dates, UUIDs and decimals, and `params.after` has them back as such.
- **Queried or sliced:** with `queried=True`, the rows are those of the page, queried with a key
  past `params.after` and `params.fetch` (one more than `limit`: it tells whether there's a next
  page). Without it, `paginate` finds the page in the whole sorted list. `descending=True` for
  newest first.
- **Signed:** with a `secret`, cursors carry an HMAC-SHA256 and a client can't make one up or
  change it; a list rotates secrets. Without it, a cursor's values are checked to be of those
  types. Either way an invalid cursor is a 400, and one longer than 1024 characters or malformed a
  422, before the endpoint runs. Cursors are base64, not encrypted: clients can read the sort key,
  so sort by what they may see.
- **Forward only:** to go back, clients keep the cursors they've used, or use
  `PaginationKeyset`, which goes both ways.

<!-- --8<-- [end:pagination-cursor] -->

### PaginationKeyset
<!-- --8<-- [start:pagination-keyset] -->

Keyset pagination in both directions: `?after=<after>` gives the page after a key and
`?before=<before>` the page before one, a key being the sort key of the item at the edge of a
page. Each page is a query through an index, from either side, as fast on the last page as on
the first, and rows added or deleted meanwhile don't shift the pages. For lists with "Previous"
and "Next", and histories scrolled both ways:

```python
from fastapi_alcazaba.pagination.keyset import KeysetPage, KeysetParams, PaginationKeyset

keyset = PaginationKeyset(secret=settings.keyset_secret)  # secrets.token_urlsafe(32)


@app.get("/products")
@cache(expire=60, tags=["products"])
async def products(params: Annotated[KeysetParams, Depends(keyset)]) -> KeysetPage[Product]:
    key = tuple_(Product.price, Product.id)
    if params.before is not None:  # the page before: the nearest rows first
        query = select(Product).where(key < params.before)
        query = query.order_by(Product.price.desc(), Product.id.desc())
    else:
        query = select(Product).order_by(Product.price, Product.id)
        if params.after is not None:
            query = query.where(key > params.after)
    rows = (await session.scalars(query.limit(params.fetch))).all()
    return keyset.paginate(rows, params, key=lambda p: (p.price, p.id), queried=True)
```

`GET /products?limit=20` answers `{"items": [...], "limit": 20, "before": null, "after":
"WzQuMiwxMl0..."}`: `after` is the key of the last item (null on the last page), and `before`
the key of the first (null on the first page). The `Link` header has the first page and, when
there's one, the page before (`before=...`) and the next (`after=...`).

- **The key** must be unique: `(price, id)`, not `price` alone, which two items may share. Its
  values can be strings, integers (up to 64 bits), floats, booleans, datetimes, dates, UUIDs
  and decimals, and `params.after` and `params.before` have them back as such.
- **Queried or sliced:** with `queried=True`, the rows are those the database returned, up to
  `params.fetch` (the limit and one more, which tells whether there's another page that way):
  past `params.after` in order, or with `params.before`, before it in the reverse order, nearest
  first, which `paginate` puts back in order. The other way there's always a page, the one the
  key came from. Without it, `paginate` finds the page in the whole sorted list.
  `descending=True` for newest first, where `after` reads `key < :after` and `before`
  `key > :before ORDER BY key`.
- **Signed:** with a `secret`, keys carry an HMAC-SHA256 and a client can't make one up or
  change it; a list rotates secrets. An invalid key is a 400 (`Invalid after`,
  `Invalid before`), and so are `after` and `before` together. Keys are base64, not encrypted:
  clients can read the sort key, so sort by what they may see.
- **Limits:** no total and no page numbers, only the first, previous and next pages. Going back
  from a page that didn't start where the first page's walk would have, the first page can be
  shorter than the limit.

<!-- --8<-- [end:pagination-keyset] -->

### PaginationIdBased
<!-- --8<-- [start:pagination-id-based] -->

The items after the last one seen, by id: `GET /messages?last_id=42&limit=20` answers
`{"items": [...], "limit": 20, "last_id": 62, "has_more": true}`, and `?last_id=62` the next page.
The id travels as it is, neither encoded nor signed, since ids are public: a client that keeps the
last one it has asks for whatever came after it. Rows added or deleted meanwhile don't shift the
pages, and the primary key finds any page as fast as the first. For syncing, chats, and lists
sorted by id:

```python
from fastapi_alcazaba.pagination.id_based import IdBasedPage, IdBasedParams, PaginationIdBased

ids = PaginationIdBased(default_limit=20, max_limit=100)  # PaginationIdBased(UUID) for UUIDs


@app.get("/messages")
@cache(expire=60, tags=["messages"])
async def messages(params: Annotated[IdBasedParams, Depends(ids)]) -> IdBasedPage[Message]:
    query = select(Message).order_by(Message.id).limit(params.fetch)
    if params.last_id is not None:
        query = query.where(Message.id > params.last_id)
    rows = (await session.scalars(query)).all()
    return ids.paginate(rows, params, key=lambda m: m.id, queried=True)
```

- **The id's type:** `int` by default, or `str` or `UUID` (`PaginationIdBased(UUID)`), which is
  the query parameter's type: a `last_id` of another type is a 422, and integers are kept to 64
  bits, as databases store them. The ids must be unique, and sort as the database sorts them.
- **Queried or sliced:** with `queried=True`, the rows are those of
  `WHERE id > :last_id ORDER BY id LIMIT :fetch`, one more than `limit` when there's a next page.
  Without it, `paginate` finds the page in the whole list, sorted by id. `descending=True` for
  newest first: the next page holds the ids smaller than `last_id`
  (`WHERE id < :last_id ORDER BY id DESC`).
- **`last_id` and `has_more`:** `last_id` is the id of the page's last item, null when it's empty
  (a client that has caught up keeps its own); `has_more` says whether there's a next page. The
  `Link` header has the first page and, while `has_more`, the next one.
- **Forward only, by one column:** to go back, clients keep the ids they've used. To sort by
  other columns, or by several, or to keep the sort key from clients, use `PaginationCursor`,
  whose cursors are opaque and can be signed.

<!-- --8<-- [end:pagination-id-based] -->

### PaginationTimeBased
<!-- --8<-- [start:pagination-time-based] -->

Pages by time: `?timestamp=2026-10-29T10:00:00Z&limit=20` gives the items older than the
timestamp, newest first (`direction="older"`, for feeds), or newer than it, oldest first
(`"newer"`, for syncing). The timestamp is plain ISO 8601, which clients can read and write,
and each page is a query through an index on the time:

```python
from fastapi_alcazaba.pagination.time_based import (
    PaginationTimeBased,
    TimeBasedPage,
    TimeBasedParams,
)

feed = PaginationTimeBased("older")


@app.get("/posts")
@cache(expire=60, tags=["posts"])
async def posts(params: Annotated[TimeBasedParams, Depends(feed)]) -> TimeBasedPage[Post]:
    query = select(Post).order_by(Post.posted_at.desc()).limit(params.fetch)
    if params.timestamp is not None:
        query = query.where(Post.posted_at < params.timestamp)
    rows = (await session.scalars(query)).all()
    return feed.paginate(rows, params, time=lambda p: p.posted_at, queried=True)
```

`GET /posts?limit=20` answers `{"items": [...], "limit": 20, "next_timestamp":
"2026-10-29T10:00:00Z"}`, the time of the page's last item, and `?timestamp=<next_timestamp>`
the next page; `next_timestamp` is null on the last one. The `Link` header has the first page
and, except on the last, the next one.

- **Ties:** the next page starts strictly past the last time, so a page never splits the items
  that share a time, or the next one would skip some. With `queried=True`, when the extra row of
  `params.fetch` (the limit and one) shares its time with the page's last items, they're left to
  the next page, which can make this one shorter; if that leaves nothing, more than `limit`
  items share one time, and `paginate` raises ValueError. Without it, `paginate` takes the rest
  of them from the whole list, past `limit` if need be.
- **Time zones:** a timestamp without one is a 422, and the parameters hold it in UTC, so the
  same instant from another zone is the same cache entry. `time` must give datetimes with one
  too.
- **Syncing:** with `"newer"`, a client keeps the time of the last item it got (at the end,
  `next_timestamp` is null) and asks for the ones after it later.
- **Limits:** when many items can share a time (imports, timestamps to the second), or rows can
  be written with a time already passed, paginate by a unique key with `PaginationKeyset`
  (the time and the id) instead.

<!-- --8<-- [end:pagination-time-based] -->

### PaginationSearchAfter
<!-- --8<-- [start:pagination-search-after] -->

The page after the sort values of the last item seen, given as a JSON array:
`?search_after=["2026-10-29T10:00:00Z",42]`. The values are plain, neither encoded nor signed:
clients read them, and may write their own to start anywhere (the events since a date). Like a
cursor, each page is a query through an index, and rows added or deleted meanwhile don't shift
the pages. For exports, logs and searches that clients drive themselves:

```python
from fastapi_alcazaba.pagination.search_after import (
    PaginationSearchAfter,
    SearchAfterPage,
    SearchAfterParams,
)

events_after = PaginationSearchAfter(sort=(datetime, int))


@app.get("/events")
@cache(expire=60)
async def events(
    params: Annotated[SearchAfterParams, Depends(events_after)],
) -> SearchAfterPage[Event]:
    query = select(Event).order_by(Event.at, Event.id).limit(params.fetch)
    if params.search_after is not None:  # (at, id) of the last event seen
        query = query.where(tuple_(Event.at, Event.id) > params.search_after)
    rows = (await session.scalars(query)).all()
    return events_after.paginate(rows, params, key=lambda e: (e.at, e.id), queried=True)
```

`GET /events?limit=20` answers `{"items": [...], "limit": 20, "search_after":
["2026-10-29T09:58:30Z", 3]}`, and `?search_after=["2026-10-29T09:58:30Z",3]` (percent-encoded
in the URL) the next page; `search_after` is null on the last one. The `Link` header has the
first page and, except on the last, the next one.

- **Declared types:** `sort` gives the types of the values, in order: `int`, `float`, `str`,
  `bool`, `datetime` (with a time zone), `date`, `UUID` and `Decimal`. `params.search_after`
  has them converted, and the page gives them as the items show them: datetimes and dates in
  ISO 8601, UUIDs and decimals as strings. A value of another type, another number of them or
  JSON that doesn't parse is a 400, and more than 1024 characters a 422, before the endpoint
  runs.
- **The key** must be unique together: `(e.at, e.id)`, not the time alone. `key` returns a
  tuple of the types of `sort`, and a value that isn't one is a TypeError.
- **Queried or sliced:** with `queried=True`, the rows are those of the page, queried past
  `params.search_after` up to `params.fetch` (the limit and one more: it tells whether there's a
  next page). Without it, `paginate` finds the page in the whole sorted list. `descending=True`
  for newest first.
- **Readable and writable:** nothing stops a client from asking for any position, so sort by
  what they may see. For a position they can't change, `PaginationCursor` seals the same key.
- **Forward only:** to go back, clients keep the values they've used.

<!-- --8<-- [end:pagination-search-after] -->

### PaginationToken
<!-- --8<-- [start:pagination-token] -->

A page token is the way to the next page and nothing else: opaque, signed, bound to the query it
was issued for, and short-lived. Clients send back what the last page gave them and never build a
page parameter of their own. For public APIs whose pagination may change behind the token without
breaking anybody, and for lists whose filters must not be swapped halfway through:

```python
from fastapi_alcazaba.pagination.page_token import PaginationToken, TokenPage, TokenParams

tokens = PaginationToken(settings.page_token_secret, ttl=600)  # secrets.token_urlsafe(32)


@app.get("/books")
@cache(expire=60)
async def books(
    params: Annotated[TokenParams, Depends(tokens)],
    genre: Genre | None = None,
) -> TokenPage[Book]:
    query = select(Book).order_by(Book.id).offset(params.offset).limit(params.fetch)
    if genre is not None:
        query = query.where(Book.genre == genre)
    rows = (await session.scalars(query)).all()
    return tokens.paginate(rows, params, queried=True)
```

`GET /books?genre=poetry&page_size=20` answers `{"items": [...], "next_page_token": "WzIwLC..."}`,
and `?genre=poetry&page_size=20&page_token=<next_page_token>` the next page; `next_page_token` is
null on the last one. The `Link` header has the next page.

- **What a token holds:** the offset of the next page, a digest of the request's other query
  parameters (all but `page_token` and `page_size`, in any order) and when it was issued, signed
  with HMAC-SHA256 (`secret` is required; a list rotates secrets). A token that a client made or
  changed, that comes with other query parameters than its own, or that is older than `ttl`
  (3600 seconds by default) is a 400 before the endpoint runs. It's base64, not encrypted:
  clients can read the offset.
- **From a database:** with `queried=True`, the rows are those of
  `LIMIT params.fetch OFFSET params.offset`, where `fetch` is `page_size` plus one: the extra row
  tells whether there's a next page. Without it, `paginate` slices the page from the whole list.
- **Limits:** `default_page_size=50, max_page_size=100`; beyond the maximum, or a malformed
  token, is a 422. `page_size` may change from one page to the next.
- **Cached apart:** the query's digest is part of `TokenParams`, so `@cache` keeps the pages of
  each query apart, even for parameters the endpoint doesn't read. A page served from the cache
  hands out the token issued when it was stored: keep `expire` well under `ttl`.
- **Its limits:** forward only, with no total and no way to jump to a page. Underneath it's an
  offset, so rows added or deleted meanwhile shift the pages, and deep pages cost what
  `OFFSET` costs; for big lists that change, sort by key with `PaginationCursor`.

<!-- --8<-- [end:pagination-token] -->

### PaginationScroll
<!-- --8<-- [start:pagination-scroll] -->

A scroll runs the query once and keeps what it found as a snapshot in the cache's backend, which
the client reads in batches: what was there when the scroll started, whatever is written
meanwhile. For exports, reports and jobs that read a whole result set while it changes, and for
queries too costly to run once per page:

```python
from fastapi_alcazaba.pagination.scroll import PaginationScroll, ScrollPage, ScrollParams

scroll = PaginationScroll(Order, secret=settings.scroll_secret)  # secrets.token_urlsafe(32)


@app.get("/orders/export")
async def export(params: Annotated[ScrollParams, Depends(scroll)]) -> ScrollPage[Order]:
    if params.scroll_id is None:  # a new scroll: query once, one row more than it keeps
        query = select(Order).where(Order.status == "paid").order_by(Order.id)
        rows = await session.scalars(query.limit(scroll.max_items + 1))
        return await scroll.start(rows.all(), params)
    return await scroll.next(params)


@app.delete("/orders/export", status_code=204)
async def stop(scroll_id: str) -> None:
    await scroll.clear(scroll_id)
```

`GET /orders/export?batch_size=100` answers the first batch,
`{"items": [...], "scroll_id": "WyJsR3Nq...", "total": 2450}`, and `?scroll_id=<scroll_id>` the
next one; `scroll_id` is null after the last. The `Link` header has the next batch.

- **A snapshot:** `start` writes the items into one entry of the backend, as JSON through their
  type (`PaginationScroll(Order, ...)`), and `next` reads them back as such. Rows added, changed
  or deleted meanwhile don't show, and no item is skipped or seen twice.
- **Kept alive while it's read:** the snapshot lasts `keep_alive` seconds (60 by default) after
  each batch; a `scroll_id` whose snapshot has expired is a 404, and the client starts again.
  `clear` deletes it as soon as the client is done. The cache's own `clear()` leaves snapshots
  alone, and with a TieredBackend they're in the shared tier only.
- **Retries are safe:** a `scroll_id` holds the snapshot's id and where the next batch starts,
  signed with HMAC-SHA256: the same request gets the same batch, and one made up or changed is a
  400 before the endpoint runs. Each request may ask for its own `batch_size`, up to
  `max_batch_size` (1000 by default; a 422 beyond).
- **Who may read it:** a `scroll_id` reads the rest of the snapshot without the endpoint's query
  running again, so whoever has one gets the results it was given for. When they depend on who
  asks, pass `owner=str(user.id)` to `start`, `next` and `clear`: a `scroll_id` is then only good
  for that user.
- **Sealed:** the snapshot is sealed with the cache's `signing_key`, or `encryption_key` when it
  holds personal data, and one that someone planted in the backend is ignored.
- **Its limits:** each batch reads the whole snapshot from the backend and writes it back to keep
  it alive, and each start keeps a new one, so `start` refuses more than `max_items` items (10,000
  by default; query one more than that and it raises ValueError) and a
  rate limit on the endpoint keeps clients from starting scrolls without end.
  The snapshot is one entry: mind the backend's limit on the size of an entry (1 MB in Memcached,
  400 KB in DynamoDB). A backend that fails is a 503, and the client retries. Don't put `@cache`
  on a scroll: each start must keep a snapshot of its own. To read a big table without keeping
  it anywhere, `PaginationStreaming` sends every row as it comes.

<!-- --8<-- [end:pagination-scroll] -->

### PaginationStreaming
<!-- --8<-- [start:pagination-streaming] -->

Every item in one response, as they come, without pages: NDJSON (`application/x-ndjson`), one
JSON value per line, which a client reads while the rest is still being queried. For exports,
for syncing everything to another system, and for scripts that would otherwise walk hundreds of
pages. It isn't a dependency: it makes the response.

```python
from fastapi.responses import StreamingResponse
from fastapi_alcazaba.pagination.streaming import PaginationStreaming

stream = PaginationStreaming(batch_size=100)


@app.get("/events")
async def events() -> StreamingResponse:
    async def fetch(after: int | None, limit: int) -> list[Event]:
        query = select(Event).order_by(Event.id).limit(limit)
        if after is not None:
            query = query.where(Event.id > after)
        return list((await session.scalars(query)).all())

    return stream.response(stream.batches(fetch, key=lambda e: e.id))
```

`GET /events` answers one event per line, `{"id":1,...}`, `{"id":2,...}`..., each hundred sent
as soon as it's read, before the next hundred are queried, until the last event.

- **`response(items)`** takes a list, a generator or an async iterable, and sends it in chunks of
  `batch_size` lines, models serialised with pydantic as FastAPI would. A synchronous generator
  is read in the thread pool, `batch_size` items at a time, so a blocking driver doesn't stop
  the event loop.
- **`batches(fetch, key=...)`** reads a big table without loading it: `fetch(None, batch_size)`,
  then `fetch(key(last row), batch_size)`, until a batch comes back shorter than `batch_size`.
  Each query finds its batch through an index, as fast as the first. `fetch` can be sync or
  async, and `key` must be unique (`(e.created_at, e.id)` rather than `e.created_at`). It
  yields the rows one by one, ready for `response()`.
- **`max_items`** ends the response there, whatever is left (None, the default, for no end).
  The source is closed then, or at its end: a generator's `finally` runs.
- **A client that disconnects** stops the stream: nothing more is read from the source, and no
  more queries are made.
- **Not cached:** `@cache` doesn't keep streams, so each request reads the database again; don't
  put it on these endpoints.
- **Its limits:** the status and headers go out before the first item, so an error halfway
  through can only cut the stream short; clients should check that they got what they expected
  (a count in another endpoint, or the last id). There's no resuming in the middle: a client
  that needs it pages by cursor instead.

<!-- --8<-- [end:pagination-streaming] -->

## Security model
<!-- --8<-- [start:security-model] -->

- **What gets cached:** `GET` requests (any call for plain functions) whose response is a 200 and
  doesn't say `Cache-Control: no-store`.
- **Per-user data:** requests are told apart by the endpoint's arguments plus the
  `vary_headers`. If a response depends on anything else (`request.state`, a header that isn't in
  `vary_headers`, the client IP…), add it to `vary_headers` or to a custom key builder.
- **The backend is trusted for availability, not for integrity:** without a `signing_key` or an
  `encryption_key`, whoever can write to it decides what your endpoints return. Use one whenever
  the backend is shared or reachable by others, and always with `PickleCoder`.
- **Confidentiality:** signed entries are readable by whoever can read the backend, its dumps
  (RDB/AOF files) and backups. When responses carry personal or sensitive data, use
  `encryption_key` instead.
- **Keys** built by the default key builder are SHA-256 digests: arguments and tokens never
  appear in the backend in clear text.
- **Fail open:** if the backend is down or an entry can't be decoded, the endpoint runs as if
  there were no cache. Rate limiters let requests through too, unless `fail_open=False`.
- **Rate limits** count clients by the address the ASGI server gives, IPv6 by /64 (configure it to
  trust your proxy, and only it), or by your `identifier`. Whoever can write to the backend can
  reset them; `clear()` can't.

Report vulnerabilities as described in [SECURITY.md](https://github.com/nonodev96/fastapi-alcazaba/blob/main/SECURITY.md).

<!-- --8<-- [end:security-model] -->

## Encrypted entries
<!-- --8<-- [start:encrypted-entries] -->

```python
FastAPICache.init(
    RedisBackend(redis),
    prefix="myapp",
    encryption_key=settings.cache_encryption_key,  # secrets.token_urlsafe(32)
)
```

Each entry is encrypted with AES-256-GCM under a key derived (HKDF-SHA256) from the secret and a
random salt, with the cache key as associated data: an entry only decrypts under its own key, and
any change to it is detected. A list of secrets rotates them: the first encrypts, all of them
decrypt. Encryption also authenticates the entries, so it replaces `signing_key` rather than
adding to it. It costs a few microseconds per entry.

<!-- --8<-- [end:encrypted-entries] -->

## Two-level cache
<!-- --8<-- [start:two-level-cache] -->

```python
from fastapi_alcazaba.backends.tiered import TieredBackend

FastAPICache.init(TieredBackend(ValkeyBackend(client), local_ttl=5), prefix="myapp")
```

`TieredBackend` keeps a small copy of each entry in the memory of every process, in front of the
shared backend: hits served from there skip the network (131 µs → 56 µs per hit against a Valkey
on the same machine; more across a real network). A local copy lives at most `local_ttl` seconds
and never longer than the shared entry.

With Redis or Valkey through redis-py/valkey-py, writes, clears and tag invalidations are also
announced on a pub/sub channel (`invalidation_channel`), so every process drops its copy at once:
a tag invalidation names the keys it deleted, also the copies of processes that only read those
entries (beyond 1000 keys, every process drops all its copies). A value read while an
invalidation arrives isn't kept, since it may be the old one. Copies are only kept while the
process is subscribed: if the subscription drops, or the client reconnects by itself, the local
level is emptied, and bypassed until it's back. The channel is read with its own timeout, so a
client `socket_timeout` doesn't end the subscription, and pinged after 5 quiet seconds: a
connection that died silently is given up within 10. With the other backends (memcached,
DynamoDB, GLIDE, a valkey-py cluster) staleness is bounded by `local_ttl`. If the shared backend
goes down, the local copies keep being served until they expire.

<!-- --8<-- [end:two-level-cache] -->

## Compression
<!-- --8<-- [start:compression] -->

```python
FastAPICache.init(RedisBackend(redis), compression="zstd")  # or "zlib", no dependencies
```

Entries of at least `compression_min_size` bytes (1 KiB) are compressed when that makes them
smaller; JSON lists shrink about 10-15 times with zstd, which means less memory in Redis/Valkey and
less network. zstd is in the standard library from Python 3.14 (`zstd` extra before). Compressed
entries are recognised when they're read whatever the setting, so it can be turned on or off at
any time; decompression is bounded by `max_value_size` (64 MiB if unset), so a compressed bomb
planted in the backend is rejected.

Compression happens before signing or encryption. With encryption, the size of an entry still
says how compressible it was: if an attacker can both read the backend and put chosen text into a
response next to a secret, that can leak information about the secret (the CRIME/BREACH family).
Leave compression off for such responses.

<!-- --8<-- [end:compression] -->

## Coders
<!-- --8<-- [start:coders] -->

`JsonCoder` (the default) stores anything FastAPI can turn into JSON, through pydantic-core
(Rust) when it can: about 8 times faster than `jsonable_encoder()` for a list of models. Annotate the
return type and cache hits come back as that type (Pydantic models, dataclasses, lists of
models…), validated straight from the stored bytes:

```python
@app.get("/products")
@cache(expire=60)
async def products() -> list[Product]: ...
```

`PickleCoder` stores any picklable object. Unpickling runs code chosen by whoever wrote the
bytes: use it only with a `signing_key` or an `encryption_key` (Alcazaba warns otherwise).

A custom coder subclasses `fastapi_alcazaba.Coder`:

```python
import orjson
from fastapi.encoders import jsonable_encoder

from fastapi_alcazaba import Coder


class ORJsonCoder(Coder):
    @classmethod
    def encode(cls, value: Any) -> bytes:
        return orjson.dumps(value, default=jsonable_encoder)

    @classmethod
    def decode(cls, value: bytes) -> Any:
        return orjson.loads(value)
```

<!-- --8<-- [end:coders] -->

## Key builders
<!-- --8<-- [start:key-builders] -->

The default key is the SHA-256 of the function's module and qualified name and the `repr()` of
its arguments, under `<prefix>:<namespace>:`. Arguments whose `repr()` changes on every call (a
database session or a service injected with `Depends`, whose `repr()` shows a memory address)
make every call a miss: Alcazaba warns about them (`UnstableCacheKeyWarning`) on the first call.
Leave them out with `exclude_params`:

```python
@app.get("/products")
@cache(expire=30, exclude_params=["session"])
def products(session: Annotated[Session, Depends(get_session)], page: int = 1) -> list[Product]: ...
```

A custom key builder receives the function, the namespace and the request:

```python
from typing import Any

from starlette.requests import Request
from starlette.responses import Response


def request_key_builder(
    func: Any,
    namespace: str = "",
    *,
    request: Request | None = None,
    response: Response | None = None,
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
) -> str:
    assert request is not None
    return ":".join(
        [
            namespace,
            request.method.lower(),
            request.url.path,
            repr(sorted(request.query_params.items())),
        ]
    )


@app.get("/")
@cache(expire=60, key_builder=request_key_builder)
async def index() -> dict[str, str]:
    return {"hello": "world"}
```

Whatever the key builder, Alcazaba appends a digest of the `vary_headers`.

<!-- --8<-- [end:key-builders] -->

## Backends
<!-- --8<-- [start:backends] -->

- **InMemoryBackend(max_size=10_000):** per process; for development, tests and single-process
  apps. Up to `max_size` cached responses, and as many keys of state (counters, revoked tokens…)
  in a store of their own.
- **RedisBackend(redis):** redis-py client, standalone or cluster, created without
  `decode_responses=True`. Works with any server that speaks the Redis protocol (Valkey, KeyDB,
  Dragonfly).
- **ValkeyBackend(valkey):** the same with valkey-py (`valkey.asyncio`, not the sync client).
- **GlideBackend(client):** Valkey GLIDE, the official Valkey client with a Rust core
  (`await GlideClient.create(...)` or `GlideClusterClient`).
- **TieredBackend(remote, local_ttl=5):** a per-process memory level in front of any of these
  (see "Two-level cache").
- **MemcachedBackend(client):** aiomcache. Can't tell the remaining time to live.
- **DynamoBackend(table_name, region, endpoint_url=None):** aiobotocore; `await backend.init()`
  first. Enable TTL on the `ttl` attribute of the table. `endpoint_url` for DynamoDB Local or
  LocalStack.

| Backend | Clear a namespace | Locks | Tags | Rate limiter |
| --- | --- | --- | --- | --- |
| InMemory | yes | (one process) | yes | (one process) |
| Redis, Valkey, GLIDE | yes (`SCAN` + `UNLINK`) | yes | yes | yes, one Lua script |
| Memcached | no | no | no | yes, adding and taking back what doesn't fit |
| DynamoDB | no | yes (conditional writes) | no | yes, a conditional write |
| Tiered | like its shared backend | like its shared backend | like its shared backend | in its shared backend |

Valkey example with valkey-py:

```python
import valkey.asyncio

from fastapi_alcazaba.backends.valkey import ValkeyBackend

client = valkey.asyncio.from_url("valkey://localhost:6379/0")
FastAPICache.init(ValkeyBackend(client), prefix="myapp", signing_key=settings.cache_signing_key)
```

<!-- --8<-- [end:backends] -->

## Development
<!-- --8<-- [start:development] -->

```shell
make deps                     # .venv (Python's venv and pip), with every extra and the dev tools
.venv/bin/pytest              # unit tests
make services-up              # Redis, Valkey, a Valkey cluster, memcached and DynamoDB Local
make test-integration         # every backend against those servers
make services-down
make mutation                 # mutation testing (slow; CI runs it weekly)
make lint                     # ruff, ruff format --check, mypy --strict, pylint
make docs                     # the documentation, built as CI does (make docs-serve to preview it)
```

Tests run in a random order (pytest-randomly; the seed is printed to reproduce a failure), and CI
fails a pull request with a benchmark more than 50 % slower than main's, compared with the others
so the speed of the runner doesn't count. See
[CONTRIBUTING.md](https://github.com/nonodev96/fastapi-alcazaba/blob/main/CONTRIBUTING.md).

<!-- --8<-- [end:development] -->

## License and credits
<!-- --8<-- [start:license-and-credits] -->

Apache License 2.0: see [LICENSE](https://github.com/nonodev96/fastapi-alcazaba/blob/main/LICENSE)
and [NOTICE](https://github.com/nonodev96/fastapi-alcazaba/blob/main/NOTICE).

An alcazaba is the walled citadel of an Andalusian city, like those of Almería, Málaga or the
Alhambra: walls around what matters, and gates that only let the right things in. The logo is one
of its walls, crowned with pointed merlons, whose horseshoe arch opens on a bolt: safe, and fast.

<!-- --8<-- [end:license-and-credits] -->
