Skip to content

korvex.middleware

Ready-made hooks. See Ready-made middleware and Sessions and CSRF.

Ready-made hooks (CORS, request IDs, rate limiting, compression, security headers, sessions, CSRF).

Guide: https://3d3n-pyc.github.io/korvex/middleware/

RequestIdFilter

Logging filter that adds request_id to records (- outside a request).

Guide: https://3d3n-pyc.github.io/korvex/middleware/#request-ids-request_idapp

RateLimitStore

Where rate limit state lives: hit decides for one request, sync or async.

Guide: https://3d3n-pyc.github.io/korvex/middleware/#your-own-store

Decision dataclass 

Decision(allowed: bool, remaining: int, retry_after: float, reset: float)

A rate limit store's answer for one request.

  • allowed: whether the request may go ahead.
  • remaining: requests left before the limit.
  • retry_after: seconds until the next request would be allowed (0 if allowed).
  • reset: seconds until the key's whole allowance is back.

MemoryStore 

MemoryStore(*, clock: Callable[[], float] = monotonic)

Rate limit state in this process's memory, one store per worker.

Algorithm: GCRA, safe with several threads.

Guide: https://3d3n-pyc.github.io/korvex/middleware/#where-the-counts-live

RedisStore 

RedisStore(client: Any = 'redis://localhost:6379/0', *, prefix: str = 'korvex:ratelimit:', fail_open: bool = True)

Rate limit state in Redis, shared by every worker and machine.

Needs pip install korvex[redis].

  • client: a URL, or a redis.asyncio.Redis you made. Connections open on first use; from a URL, each event loop gets its own client.
  • prefix: put before each key.
  • fail_open: if Redis is unreachable, allow requests (default) or refuse them with a 429; the error is logged either way.

Same algorithm as MemoryStore, run atomically on Redis's clock. Guide: https://3d3n-pyc.github.io/korvex/middleware/#where-the-counts-live

Session 

Session(data: Optional[dict[str, Any]] = None)

The data kept for one browser between requests, saved in a signed cookie.

Use it like a dict. Values must be JSON-serialisable. Setting, deleting or clearing a key sets modified; a change inside a nested value is not seen, so set session.modified = True yourself after it. Guide: https://3d3n-pyc.github.io/korvex/sessions/

Starts with data, not modified.

clear 

clear() -> None

Removes everything and marks the session modified.

pop 

pop(*args: Any) -> Any

Removes a key and returns its value, marking the session modified.

popitem 

popitem() -> tuple[str, Any]

Removes the last item, marking the session modified.

setdefault 

setdefault(key: str, default: Any = None) -> Any

Returns a key's value, setting it first if missing.

update 

update(*args: Any, **kwargs: Any) -> None

Updates from a mapping or pairs, marking the session modified.

cors 

cors(app: App, *, allow_origins: Sequence[str] | str, allow_methods: Sequence[str] = ALL_METHODS, allow_headers: Sequence[str] | str = (), allow_credentials: bool = False, expose_headers: Sequence[str] = (), max_age: int = 600) -> None

Lets browsers call the app from other origins (CORS).

  • allow_origins: origins like "https://example.com", or "*".
  • allow_methods: methods a cross-origin request may use.
  • allow_headers: request headers allowed beyond the basic ones, or "*".
  • allow_credentials: whether cookies may be sent. With "*" origins, the request's origin is sent back instead of *, as browsers require.
  • expose_headers: response headers the page's code may read.
  • max_age: seconds a preflight answer may be cached.

Guide: https://3d3n-pyc.github.io/korvex/middleware/#cors-cors-app-allow-origins

request_id 

request_id(app: App, *, header: str = 'X-Request-ID', trust_incoming: bool = True) -> None

Gives every request an ID, sent back in the response and put in the logs.

  • header: the header the ID is read from and sent back in.
  • trust_incoming: reuse a safe incoming ID, else make a new one.

The built-in server's access logs show it too. Guide: https://3d3n-pyc.github.io/korvex/middleware/#request-ids-request_idapp

current_request_id 

current_request_id() -> str | None

The ID of the request being handled, if request_id() is set up.

rate_limit 

rate_limit(app: App, *, requests: int, per: float, key: Callable[[Connection], Optional[Hashable]] = _client_address, store: RateLimitStore | None = None, headers: bool = True) -> None

Allows each key requests requests per per seconds; beyond, a 429 with Retry-After.

  • key(request): who the limit applies to, the client's address by default (unknown addresses share one limit); None exempts a request.
  • store: where the state lives, a MemoryStore by default.
  • headers: send RateLimit-Limit, RateLimit-Remaining and RateLimit-Reset.

Runs as a before hook: hooks declared before it (like cors()) run first. Guide: https://3d3n-pyc.github.io/korvex/middleware/#rate-limiting-rate_limitapp-requests100-per60

compress 

compress(app: App, *, minimum_size: int = 500, encodings: Sequence[str] | None = None, levels: dict[str, int] | None = None, content_types: Sequence[str] = COMPRESSIBLE) -> None

Compresses responses for clients that accept it.

  • encodings: in order of preference; ("zstd", "br", "gzip") by default (br is left out of the default on free-threaded Python, but can be asked for). Unavailable ones are skipped.
  • minimum_size: smaller bodies are sent as they are; streams are always compressed.
  • levels: per encoding, like {"gzip": 9}.
  • content_types: prefixes like text/ or suffixes like +json.

Call it first: after hooks run last-declared first. Compressed responses get Vary: Accept-Encoding. Don't compress pages mixing a secret with client input (BREACH), over HTTPS too. Guide: https://3d3n-pyc.github.io/korvex/middleware/#compression-compressapp

security_headers 

security_headers(app: App, *, hsts: int | None = 31536000, hsts_subdomains: bool = False, hsts_preload: bool = False, content_security_policy: str | None = None, frame_options: str | None = 'DENY', content_type_options: bool = True, referrer_policy: str | None = 'strict-origin-when-cross-origin', cross_origin_opener_policy: str | None = 'same-origin', permissions_policy: str | None = None) -> None

Adds the usual security headers to every response; one a response sets keeps its own.

  • hsts: Strict-Transport-Security max-age in seconds, None to leave it out; sent only when request.scheme is https (set by a trusted proxy); raises ValueError if negative.
  • hsts_subdomains, hsts_preload: add those directives.
  • content_security_policy: off by default, like "default-src 'self'".
  • frame_options: X-Frame-Options.
  • content_type_options: send X-Content-Type-Options: nosniff.
  • referrer_policy, cross_origin_opener_policy, permissions_policy: those headers; None leaves one out.

Guide: https://3d3n-pyc.github.io/korvex/middleware/#security-headers-security_headersapp

sessions 

sessions(app: App, *, secret_key: Union[str, bytes, Sequence[Union[str, bytes]]], max_age: int = 14 * 24 * 3600, cookie: str = 'session', secure: bool = True, samesite: str = 'lax', path: str = '/', domain: Optional[str] = None) -> None

Keeps a Session per browser in a signed, HttpOnly cookie; read it with current_session().

  • secret_key: at least 32 bytes; a list rotates keys (the first signs, all verify).
  • max_age: seconds a session stays valid.
  • cookie, secure, samesite, path, domain: the cookie's name and attributes.

Only a session that was read and changed is sent; the cookie is limited to 4096 bytes. Guide: https://3d3n-pyc.github.io/korvex/sessions/

current_session 

current_session() -> Session

The session of the request being handled, read from its cookie on first use.

A missing, expired or tampered cookie gives an empty session. Raises RuntimeError outside a request or without sessions(app, ...). In a WebSocket it reads the handshake's cookie; changes cannot be saved. Guide: https://3d3n-pyc.github.io/korvex/sessions/

flash 

flash(message: str, category: str = 'info') -> None

Keeps a message in the session to show on the next page.

  • message: the text.
  • category: a label such as "info" or "error".

Guide: https://3d3n-pyc.github.io/korvex/sessions/

get_flashed_messages 

get_flashed_messages() -> list[tuple[str, str]]

Returns the (category, message) pairs kept by flash() and forgets them.

Guide: https://3d3n-pyc.github.io/korvex/sessions/

csrf 

csrf(app: App, *, trusted_origins: Iterable[str] = (), exempt: Iterable[str] = ()) -> None

Refuses cross-site requests that change things, without tokens (CSRF).

It trusts what the browser sends (Sec-Fetch-Site, else Origin against Host). GET, HEAD and OPTIONS pass; WebSocket handshakes are always checked. Requests with neither header pass.

  • trusted_origins: other origins allowed, like "https://admin.example.com".
  • exempt: paths to skip: exact ("/hook") or ending in / for a prefix ("/webhooks/").

Guide: https://3d3n-pyc.github.io/korvex/sessions/