# Korvex

> A fast Python web framework, with a router and an HTTP server written in Rust.

Korvex is a fast Python web framework (ASGI), with a router and an HTTP server written in Rust. Handlers are plain functions whose parameters say what they read; validation uses msgspec or pydantic. Python 3.9 to 3.14, free-threaded 3.14t included. Korvex 1.0 is in release candidates: install with `pip install --pre korvex` (plain `pip install korvex` gives 0.17, an older API); the API rc1 froze won't break.


# Start

# Korvex

A fast Python web framework, with a router and an HTTP server written in Rust. You write plain Python functions; Korvex calls them for the right requests, checks their input, and documents your API. Use it for JSON APIs, HTML sites and WebSockets where speed matters and plain functions are enough.

New to web APIs? Start with the [tutorial](https://3d3n-pyc.github.io/korvex/tutorial/index.md).

```
pip install --pre korvex
```

## A first app

```
from korvex import App, Response

app = App()


@app.get("/hello/{name}")
def hello(name: str, polite: bool = False):
    return Response.json({"message": f"Good day, {name}" if polite else f"Hi {name}"})


if __name__ == "__main__":
    app.run()
```

Run `python main.py`, then open <http://127.0.0.1:8000/hello/Ada?polite=true>.

- `name` comes from the path, `polite` from the query string, turned into a `bool`: a parameter's name and type say where it comes from and what it must be.
- A wrong `polite` (`?polite=maybe`) gets a 422 error, with no code of yours.
- `Response.json()` sends JSON. With a model (msgspec or pydantic) and a return annotation, return the model: see [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md).
- `app.docs()` adds a Swagger UI page at `/docs`.

Korvex 1.0 is in release candidates: what rc1 froze won't break; features added since can still change until 1.0.0. Install with `pip install --pre korvex`. Every change is in the [changelog](https://github.com/3d3n-pyc/korvex/blob/main/CHANGELOG.md).

## The guide

Read in order the first time; each page stands alone after.

1. [Routes](https://3d3n-pyc.github.io/korvex/routes/index.md): paths, parameters, groups, hosts, links between routes, static files.
1. [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md): typed parameters, models, headers, cookies and forms.
1. [Requests and responses](https://3d3n-pyc.github.io/korvex/requests-and-responses/index.md): what a handler gets and returns, streams, Server-Sent Events, background tasks.
1. [Templates](https://3d3n-pyc.github.io/korvex/templates/index.md): HTML pages, filters, escaping, hashed static URLs.
1. [Sessions and CSRF](https://3d3n-pyc.github.io/korvex/sessions/index.md): signed-cookie sessions, flash messages, CSRF protection.
1. [Dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md): code shared by routes, like signing in.
1. [Errors and hooks](https://3d3n-pyc.github.io/korvex/errors-and-hooks/index.md): errors, code that runs around every request, startup and shutdown.
1. [Ready-made middleware](https://3d3n-pyc.github.io/korvex/middleware/index.md): CORS, request IDs, rate limiting, compression, security headers.
1. [WebSockets](https://3d3n-pyc.github.io/korvex/websockets/index.md)
1. [OpenAPI and security](https://3d3n-pyc.github.io/korvex/openapi/index.md): the API's description, Swagger UI, sign-in schemes.
1. [Testing](https://3d3n-pyc.github.io/korvex/testing/index.md)
1. [The built-in server](https://3d3n-pyc.github.io/korvex/server/index.md): running in production, workers and threads.
1. [Deployment](https://3d3n-pyc.github.io/korvex/deployment/index.md): behind nginx or Caddy, in Docker, on AWS Lambda or Vercel.
1. [Performance](https://3d3n-pyc.github.io/korvex/performance/index.md): the benchmarks, and how to reproduce them.

Terms you don't know are in the [glossary](https://3d3n-pyc.github.io/korvex/glossary/index.md).

The [reference](https://3d3n-pyc.github.io/korvex/reference/app/index.md) lists every class, function and option, from the code's own docstrings.

Also:

- [The standalone router](https://3d3n-pyc.github.io/korvex/router/index.md): Korvex's Rust router on its own.
- [Examples](https://github.com/3d3n-pyc/korvex/tree/main/examples): from a few lines to a complete app.

# Your first API

You will build a small to-do API: list, read, add and filter tasks. You need basic Python and nothing else. Linked words, like [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler), are explained in the [glossary](https://3d3n-pyc.github.io/korvex/glossary/index.md).

## 1. Install

A virtual environment keeps this project's packages apart from the others. Create one, then install Korvex, msgspec (it checks your data) and pytest:

```
mkdir todo && cd todo
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install --pre korvex msgspec pytest
```

`--pre` is needed while Korvex 1.0 is a release candidate. The built-in server runs on Linux and macOS; on Windows, `pip install uvicorn` and run `uvicorn main:app` (any [ASGI](https://3d3n-pyc.github.io/korvex/glossary/#asgi) server works).

## 2. A first route

A [route](https://3d3n-pyc.github.io/korvex/glossary/#route) links a method and a path (`GET /tasks`) to a [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler): a plain function Korvex calls for that request. It returns a `Response`; `Response.json` sends JSON. Create `main.py`:

```
from korvex import App, Response

app = App()

TASKS = {
    1: {"id": 1, "title": "Install Korvex", "done": True},
    2: {"id": 2, "title": "Build a to-do API", "done": False},
}


@app.get("/tasks")
def list_tasks():
    return Response.json(list(TASKS.values()))


if __name__ == "__main__":
    app.run()
```

Run it with `korvex run main:app` (`main` is the file, `app` the variable) or `python main.py`, then open <http://127.0.0.1:8000/tasks>: you see your two tasks. Stop with Ctrl+C.

## 3. One task, and a 404

`{id:int}` in a path is a [path parameter](https://3d3n-pyc.github.io/korvex/glossary/#path-parameter): the part of the URL that changes. Naming the handler's parameter `id` with type `int` makes Korvex pass the number, and refuse `/tasks/abc`. For an unknown task, `raise HTTPException(404, ...)` sends a 404 with a JSON message.

Import `HTTPException` too, and add this below `list_tasks`:

```
@app.get("/tasks/{id:int}")
def get_task(id: int):
    if id not in TASKS:
        raise HTTPException(404, "no such task")
    return Response.json(TASKS[id])
```

Restart the server: `/tasks/9` now shows `{"detail": "no such task"}`.

## 4. A model for the request body

So far we sent dicts and trusted them. A [model](https://3d3n-pyc.github.io/korvex/glossary/#model) says what a task looks like: names and types. Korvex uses it two ways. A handler parameter typed with a model is read from the JSON [request body](https://3d3n-pyc.github.io/korvex/glossary/#request-body) and checked ([validation](https://3d3n-pyc.github.io/korvex/glossary/#validation)). A return annotation (`-> Task`) means "send this model as JSON". Here, a new task needs only a `title`; Korvex fills `id` and `done`. Make `TASKS` a `dict[int, Task]`, and have `list_tasks` and `get_task` return them directly, with `-> list[Task]` and `-> Task` and no `Response.json`: the complete file is below. Add the models and a create route:

```
import msgspec

from korvex import App, HTTPException


class NewTask(msgspec.Struct):
    title: str


class Task(msgspec.Struct):
    id: int
    title: str
    done: bool = False


@app.post("/tasks", status=201)
def create_task(body: NewTask) -> Task:
    task = Task(id=len(TASKS) + 1, title=body.title)
    TASKS[task.id] = task
    return task
```

`status=201` means "created". Try `curl -X POST localhost:8000/tasks -d '{"title": "Write tests"}'`: it returns the new task. With `{"title": 3}` (a title must be text), Korvex answers `422` before your code runs, and says where and why:

```
{"detail": [{"loc": ["body", "title"], "msg": "Expected `str`, got `int`", "type": "validation"}]}
```

## 5. A query parameter

The [query string](https://3d3n-pyc.github.io/korvex/glossary/#query-string) is the part after `?` in `/tasks?done=true`. A parameter that is not in the path and has a simple type (`str`, `int`, `bool`...) is read from it. `Optional[bool] = None` means "optional; absent means no filter". Replace `list_tasks`, and add `from typing import Optional` at the top:

```
@app.get("/tasks")
def list_tasks(done: Optional[bool] = None) -> list[Task]:
    return [t for t in TASKS.values() if done is None or t.done == done]
```

`/tasks?done=true` lists finished tasks; `?done=maybe` is a 422.

## 6. A test

A test calls your app without starting a server, so it runs in milliseconds. `TestClient` is a fake browser; a response has `status` and `json()`. Save `test_main.py` below, then run `pytest`.

## 7. The complete files

**main.py**

```
from typing import Optional

import msgspec

from korvex import App, HTTPException


class NewTask(msgspec.Struct):
    title: str


class Task(msgspec.Struct):
    id: int
    title: str
    done: bool = False


app = App()

TASKS: dict[int, Task] = {
    1: Task(id=1, title="Install Korvex", done=True),
    2: Task(id=2, title="Build a to-do API"),
}


@app.get("/tasks")
def list_tasks(done: Optional[bool] = None) -> list[Task]:
    return [t for t in TASKS.values() if done is None or t.done == done]


@app.get("/tasks/{id:int}")
def get_task(id: int) -> Task:
    if id not in TASKS:
        raise HTTPException(404, "no such task")
    return TASKS[id]


@app.post("/tasks", status=201)
def create_task(body: NewTask) -> Task:
    task = Task(id=len(TASKS) + 1, title=body.title)
    TASKS[task.id] = task
    return task


if __name__ == "__main__":
    app.run()
```

**test_main.py**

```
from korvex.testing import TestClient

from main import app


def test_tasks():
    with TestClient(app) as client:
        done = client.get("/tasks", query={"done": "true"}).json()
        assert done and all(task["done"] for task in done)
        created = client.post("/tasks", json={"title": "Write tests"})
        assert created.status == 201
        assert client.get(f"/tasks/{created.json()['id']}").json()["title"] == "Write tests"


def test_errors():
    with TestClient(app) as client:
        assert client.get("/tasks/999").status == 404
        assert client.post("/tasks", json={"title": 3}).status == 422
        assert client.get("/tasks", query={"done": "maybe"}).status == 422
```

## Where to go next

- [Routes](https://3d3n-pyc.github.io/korvex/routes/index.md): groups, hosts, links between routes, static files.
- [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md): headers, cookies, forms, your own 422 reply.
- [Testing](https://3d3n-pyc.github.io/korvex/testing/index.md): async tests, WebSockets, replacing dependencies.
- [Deployment](https://3d3n-pyc.github.io/korvex/deployment/index.md): workers, nginx, Docker.
- [OpenAPI](https://3d3n-pyc.github.io/korvex/openapi/index.md): `app.docs()` adds a page at `/docs` to try each route.
- [Glossary](https://3d3n-pyc.github.io/korvex/glossary/index.md): every term used in this guide.

# Glossary

Short meanings for the words the guide uses, each with where Korvex covers it.

## ASGI

The standard way a Python web server talks to a web app: the server calls the app with a request and the app sends back a response. Because of it, one app can run on many servers.

In Korvex: an `App` is an ASGI app, so the built-in server, uvicorn, Granian or a serverless platform can run it. See [The built-in server](https://3d3n-pyc.github.io/korvex/server/index.md).

## Handler

The function that answers a request: it gets what it asks for as parameters and returns the response. It can be sync (`def`) or async (`async def`).

In Korvex: `@app.get("/users")` above a function. See [Routes](https://3d3n-pyc.github.io/korvex/routes/index.md) and [Requests and responses](https://3d3n-pyc.github.io/korvex/requests-and-responses/#what-a-handler-gets).

## Route

A rule that links a method (`GET`, `POST`...) and a URL path to a handler. Korvex picks the route that matches each request.

In Korvex: `@app.get`, `@app.post` and the like. See [Routes](https://3d3n-pyc.github.io/korvex/routes/index.md).

## Path parameter

A changing part of the URL that the handler receives, like the `7` in `/users/7`.

In Korvex: `"/users/{id:int}"`, then `def get_user(id: int)`. See [Path parameters](https://3d3n-pyc.github.io/korvex/routes/#path-parameters).

## Query string

What follows the `?` in a URL, like `?page=2&tag=a`. It carries optional settings of the request.

In Korvex: a parameter of a simple type (`page: int = 1`) comes from it, and `request.query` reads it. See [Where a parameter comes from](https://3d3n-pyc.github.io/korvex/validation/#where-a-parameter-comes-from).

## Request body

The data a client sends after the headers, usually with `POST`, `PUT` or `PATCH`: JSON, a form or a file. It is read from the connection, so a very large one is limited.

In Korvex: a model parameter reads the JSON body, and `await request.body()`, `.json()` and `.form()` read it by hand. See [The request](https://3d3n-pyc.github.io/korvex/requests-and-responses/#the-request).

## Model

A class that describes the shape of some data (`name: str`, `age: int`). Korvex uses one to check an incoming body and to describe the API. It can be a msgspec struct, a pydantic model, a dataclass or a `TypedDict`. To choose: msgspec is the fastest, pydantic is the most common. Install one (`pip install msgspec` or `pip install pydantic`). Korvex needs neither, unless you validate bodies (a dataclass or `TypedDict` is checked by whichever is installed).

In Korvex: a `msgspec.Struct` or a pydantic `BaseModel` as a parameter or a return annotation. See [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md).

## Validation

Checking that incoming data is what you expect (a number where a number is due, a required field present) before your code runs. Without it, every handler must check by hand.

In Korvex: type annotations do it, and a wrong value gets a 422 error with no code of yours. See [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md) and its [Errors](https://3d3n-pyc.github.io/korvex/validation/#errors).

## Dependency

A function whose result a handler receives, for code that several routes share: who is signed in, a database connection.

In Korvex: `Depends(func)`. See [Dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md).

## Hook

A function Korvex calls at a set moment, so you can act on every request without editing each handler.

In Korvex: `@app.before_request`, `@app.after_request`, `@app.on_startup`, `@app.on_shutdown`. See [Errors and hooks](https://3d3n-pyc.github.io/korvex/errors-and-hooks/#hooks-around-every-request).

## Middleware

Code that sits between the server and your handlers and sees every request and response: logging, compression, rate limiting.

In Korvex: ready-made ones are functions like `cors(app, ...)` that register a hook on the app. See [Ready-made middleware](https://3d3n-pyc.github.io/korvex/middleware/index.md).

## Lifespan

The life of the app from start to stop. Things you open once, like a database pool, are created at its start and closed at its end, not per request.

In Korvex: `@app.on_startup` and `@app.on_shutdown`; the state they keep goes in `app.local`. See [Startup and shutdown](https://3d3n-pyc.github.io/korvex/errors-and-hooks/#startup-and-shutdown).

## Event loop

The loop that runs async code: it switches between tasks whenever one waits (for the network, a database). A sync handler runs on the event loop too, so blocking code in it (a slow sync database call, `time.sleep`) holds up every other request in that worker. Use async libraries, or move the blocking call to a thread with `await asyncio.to_thread(...)`.

In Korvex: the built-in server runs one loop per worker, or per thread with `--threads`. See [Threads](https://3d3n-pyc.github.io/korvex/server/#threads).

## Worker

A process that runs your app and answers requests. Several workers share one port, so the app uses several CPU cores, and one that crashes is restarted.

In Korvex: `korvex run main:app --workers 3` or `app.run(workers=3)`. See [Options](https://3d3n-pyc.github.io/korvex/server/#options).

## Reverse proxy

A server (nginx, Caddy, Traefik) in front of your app. It takes the traffic from the internet, handles HTTPS, and forwards requests to the app.

In Korvex: run the built-in server behind one, and tell it which proxies to trust with `--forwarded-allow-ips`. See [Deployment](https://3d3n-pyc.github.io/korvex/deployment/#1-built-in-server-behind-a-reverse-proxy).

## root_path

The URL prefix an app is served under, like `/api`, when a proxy strips it before forwarding or when an app is mounted. The app needs it to build correct links.

In Korvex: `--root-path` on the server, `request.root_path` in a handler, and `app.mount(prefix, other_app)`. See [Options](https://3d3n-pyc.github.io/korvex/server/#options) and [Mounting another app](https://3d3n-pyc.github.io/korvex/routes/#mounting-another-app).

## CORS

Cross-Origin Resource Sharing: browsers block a page on one site from calling the API of another unless that API says it may. It is a browser rule; curl and other servers are not affected.

In Korvex: `cors(app, allow_origins=[...])`. See [CORS](https://3d3n-pyc.github.io/korvex/middleware/#cors-corsapp-allow_origins).

## CSRF

Cross-site request forgery: a page on another site makes the visitor's browser send a request to yours, with the visitor's cookies. It matters for any app that signs users in with a cookie.

In Korvex: `csrf(app)` refuses such requests with a 403. See [CSRF](https://3d3n-pyc.github.io/korvex/sessions/#csrf-csrfapp).

## Session

Data kept for one visitor between requests, such as who is signed in. The browser holds a cookie that links its requests together.

In Korvex: `sessions(app, secret_key=...)` stores it in a signed cookie, and `current_session()` reads it. See [What a session is](https://3d3n-pyc.github.io/korvex/sessions/#what-a-session-is).

## ETag

A label (a fingerprint) the server puts on a file's content. The browser sends it back, and if the file is unchanged the server answers 304 with no body, which saves bandwidth.

In Korvex: static files get an `ETag` and a `Last-Modified` on their own. See [Static files](https://3d3n-pyc.github.io/korvex/routes/#static-files).

## Range request

A request for only part of a file (`Range: bytes=0-999`). It allows resumable downloads and seeking in a video.

In Korvex: static files and `Response.file(path, request=request)` answer it with a 206. See [Static files](https://3d3n-pyc.github.io/korvex/routes/#static-files).

## WebSocket

A connection that stays open so the client and the server can send messages to each other at any time, like a chat.

In Korvex: `@app.websocket("/chat")` and a `WebSocket` parameter. See [WebSockets](https://3d3n-pyc.github.io/korvex/websockets/index.md).

## Server-Sent Events

A stream of events from the server to the browser over one HTTP response, one way only. Simpler than a WebSocket when the client only listens.

In Korvex: `Response.sse(events())`, with a generator that yields `ServerSentEvent`s. See [Server-Sent Events](https://3d3n-pyc.github.io/korvex/requests-and-responses/#server-sent-events).

## Background task

Work started during a request but run after the response is sent, such as sending an email, so the client does not wait for it.

In Korvex: `tasks.add(func, ...)`; a coroutine function is awaited and a plain function runs in a thread. See [Background tasks](https://3d3n-pyc.github.io/korvex/requests-and-responses/#background-tasks).
# Guide

# Routes

Use this when you declare which URLs your app answers, and with which handler.

```
from korvex import App

app = App()


@app.get("/users/{id:int}")
def get_user(id: int):
    ...


@app.post("/users")
def create_user():
    ...
```

`@app.get`, `post`, `put`, `patch`, `delete`, `head` and `options` declare a [route](https://3d3n-pyc.github.io/korvex/glossary/#route) for one method. A function can be sync or async.

Note

A plain `def` [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler) runs on the [event loop](https://3d3n-pyc.github.io/korvex/glossary/#event-loop), so a blocking call in it (a synchronous database driver, `requests`, `time.sleep`) holds up every other request of the [worker](https://3d3n-pyc.github.io/korvex/glossary/#worker). Prefer `async def` with async libraries, or `await asyncio.to_thread(...)` for the blocking part.

## Path parameters

A `{name}` segment is a [path parameter](https://3d3n-pyc.github.io/korvex/glossary/#path-parameter): it captures part of the path, and the handler gets it by name:

| Segment       | Matches                              | Gives  |
| ------------- | ------------------------------------ | ------ |
| `users`       | exactly `users`                      |        |
| `{id}`        | any one segment                      | `str`  |
| `{id:int}`    | an integer (`42`, `-7`)              | `int`  |
| `{id:uuid}`   | a UUID                               | `UUID` |
| `{post:slug}` | what a constraint you defined allows | `str`  |
| `*path`       | the rest of the path (last segment)  | `str`  |

A path that doesn't fit gets a 404. When several routes could match, the most precise wins: a literal segment, then constrained parameters, then `{id}`, then `*path`.

An annotation converts a value further: `def on_day(day: date)` for `/days/{day}` gets a `date`, and a value that isn't one gets a 422.

### Your own constraints

```
from korvex import App

app = App()
app.constraint("slug", r"[a-z0-9]+(?:-[a-z0-9]+)*")


@app.get("/posts/{post:slug}")  # /posts/hello-world, not /posts/Hello
def show_post(post: str): ...
```

The whole segment must match the regex. Define a constraint before the routes that use it. Regexes are matched in Rust, in linear time (no look-around or backreferences).

## Groups

A group shares a prefix, and [dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md), between routes:

```
from korvex import App, Depends

app = App()


def signed_in(): ...  # your check; raise HTTPException(401) to refuse


api = app.group("/api", dependencies=[Depends(signed_in)])


@api.get("/tasks")  # GET /api/tasks, after signed_in
def list_tasks(): ...


admin = api.group("/admin")  # /api/admin, signed_in too
```

A group also takes `tags=`, `deprecated=`, `responses=` and `include_in_schema=` for [OpenAPI](https://3d3n-pyc.github.io/korvex/openapi/index.md); a route's own add to them.

## Hosts

A host group answers only for one host name:

```
from korvex import App

app = App()
api = app.host("api.example.com")
tenants = app.host("{tenant}.example.com")


@tenants.get("/")  # acme.example.com: tenant="acme"
def home(tenant: str): ...
```

- `{tenant}` takes one whole label of the host name, and reaches the handlers and dependencies that ask for it, like a path parameter.
- The first host pattern that fits owns the request: its routes answer, then the routes declared without a host. Patterns without a parameter are tried first.

## Links to a route

```
from korvex import App, Request, Response

app = App()


@app.get("/users/{id:int}")
def get_user(id: int): ...


@app.get("/profile")
def profile(request: Request):
    return Response.text(request.url_for("get_user", id=42))  # https://example.com/users/42


app.url_for("get_user", id=42)  # "/users/42"
```

- A route is known by its function's name, or `name=` on its decorator: `@app.get("/users/{id:int}", name="user")`.
- Values are encoded and checked: one that doesn't fit the route (`id="abc"`) raises an error.
- `request.url_for()` gives the full URL, with the route's own host for a host route. It uses the request's `Host` header, which the client sends: for a link in an email, build it from your own base URL and `app.url_for()`.
- Two routes can share a function's name; `url_for()` then asks you for a `name=` to tell them apart.

## Static files

```
from korvex import App

app = App()
app.static("/static", "public")  # /static/css/app.css: public/css/app.css
app.url_for("static", path="css/app.css")
```

A path that leaves the directory gets a 404. With several directories, give each a `name=`.

Files are served as conditional requests:

- Every file gets an [`ETag`](https://3d3n-pyc.github.io/korvex/glossary/#etag) and a `Last-Modified`. A request that sends `If-None-Match` or `If-Modified-Since` for an unchanged file gets a 304, with no body.
- [`Range: bytes=...`](https://3d3n-pyc.github.io/korvex/glossary/#range-request) (one range) gets a 206 with that part of the file, for resumable downloads and for seeking in a video. A range past the end of the file gets a 416. `If-Range` is honored; several ranges at once get the whole file.
- A URL without a hash is sent with `Cache-Control: no-cache`: the browser asks each time, and an unchanged file costs a cheap 304. A hashed URL (`app.static_url("css/app.css")`) stays `public, max-age=31536000, immutable`.

In your own handler, `Response.file(path, request=request)` does the same: see [Streams and files](https://3d3n-pyc.github.io/korvex/requests-and-responses/#streams-and-files).

## Mounting another app

```
from korvex import App, Request, Response

admin_app = App()  # or a Starlette or FastAPI app, or Django's ASGI app
app = App()
app.mount("/admin", admin_app, name="admin")

app.url_for("admin")  # "/admin/"
app.url_for("admin", path="users")  # "/admin/users"


@app.get("/link")
def link(request: Request):
    return Response.text(request.url_for("admin", path="users"))  # https://example.com/admin/users
```

`mount(prefix, asgi_app)` hands every request under `prefix` to any [ASGI](https://3d3n-pyc.github.io/korvex/glossary/#asgi) app: a Korvex `App`, a Starlette or FastAPI app, Django's ASGI app, or a plain `async def (scope, receive, send)`.

- The app gets the request with [`root_path`](https://3d3n-pyc.github.io/korvex/glossary/#root_path) set to the prefix (plus the server's own `root_path`, if any). `path` still holds the whole path: ASGI says an app strips its `root_path` itself, as a Korvex `App` does, so `/admin/users` is `/users` for `admin_app`.
- `/admin` answers 308 to `/admin/`, keeping the query string. The prefix starts with `/`, is not `/` itself and holds no `{}` or `*`.
- A route of the parent app wins: with `@app.get("/admin/health")`, that route answers and the rest of `/admin/...` goes to the mount.
- The mounted app's [lifespan](https://3d3n-pyc.github.io/korvex/glossary/#lifespan) runs with the parent's startup and shutdown, and the state it sets reaches its requests. If its startup fails, the parent's does.
- On the built-in server, a mounted app that waits for the client to leave (`http.disconnect`) learns it when its response ends; for long polling, stream the response.
- [WebSockets](https://3d3n-pyc.github.io/korvex/glossary/#websocket) under the prefix go to the mounted app too.
- The mount is not in the parent's [OpenAPI](https://3d3n-pyc.github.io/korvex/openapi/index.md) document.
- `name=` is for `url_for()`: with `path=`, a value that is empty, `.` or `..` in a part raises an error. Other parameters are refused.

Warning

The parent's before [hooks](https://3d3n-pyc.github.io/korvex/glossary/#hook) run before the mounted app, so [`csrf()`](https://3d3n-pyc.github.io/korvex/glossary/#csrf), `rate_limit()` and [`sessions()`](https://3d3n-pyc.github.io/korvex/glossary/#session) protect it (see [Middleware](https://3d3n-pyc.github.io/korvex/middleware/index.md)). Its after hooks and exception handlers do not run: [`cors()`](https://3d3n-pyc.github.io/korvex/glossary/#cors), `security_headers()`, `compress()` and the request ID header don't apply to the mounted app. Configure them in the mounted app. Only the built-in server's own maximum body size applies, not a route's `max_body=`.

## HEAD, 404 and 405

- A `HEAD` request is answered by the `GET` route, without the body.
- An unknown path gets a 404; a known path with the wrong method a 405, with an `Allow` header.

## Route options

Every decorator also takes:

- `name=`: for [`url_for()`](#links-to-a-route).
- `dependencies=[...]`: see [Dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md).
- `status=`, `tags=`, `deprecated=`, `responses=`, `include_in_schema=`: see [OpenAPI](https://3d3n-pyc.github.io/korvex/openapi/index.md).
- `max_body=`: the largest body the [built-in server](https://3d3n-pyc.github.io/korvex/server/index.md) accepts for this route.

Next: [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md).

# Validation

Use this when you want Korvex to read, check and convert the input of a [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler) from its annotations.

Annotations say where a parameter comes from, and what it must be. [Models](https://3d3n-pyc.github.io/korvex/glossary/#model) can be [msgspec](https://jcristharif.com/msgspec/) structs, [pydantic](https://docs.pydantic.dev/) models, or plain dataclasses and `TypedDict`s; Korvex needs none of these libraries to import, only to validate.

```
import msgspec

from korvex import App

app = App()


class NewUser(msgspec.Struct):
    name: str
    age: int


class User(msgspec.Struct):
    id: int
    name: str


@app.post("/users", status=201)
def create_user(body: NewUser, notify: bool = False) -> User:
    return User(id=1, name=body.name)
```

## Where a parameter comes from

| Parameter                                                                                        | Comes from                                                                                       |
| ------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------ |
| a model (struct, pydantic model, dataclass, `TypedDict`), or a list of them (`body: NewUser`)    | the JSON [body](https://3d3n-pyc.github.io/korvex/glossary/#request-body)                        |
| named like a [path parameter](https://3d3n-pyc.github.io/korvex/glossary/#path-parameter) (`id`) | the path                                                                                         |
| `str`, `int`, `float`, `bool`, `UUID`, `date`, `datetime`, or an `Optional` or `list` of them    | the [query string](https://3d3n-pyc.github.io/korvex/glossary/#query-string)                     |
| `Header()`, `Cookie()`, `Form()`, `Query(alias=...)`                                             | there                                                                                            |
| `Depends(func)`                                                                                  | what `func` returns: see [Dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md) |
| named `request`, or annotated `Request`                                                          | the request itself                                                                               |

- A default makes a parameter optional. Dates are ISO 8601.
- Anything else is refused with a message when the route is declared, so a mistake shows at import.
- When a field is repeated (`?tag=a&tag=b`), a single value takes the first, and a `list[str]` takes them all.

## Dataclasses and TypedDict

A `@dataclass` or a `TypedDict` (or a list or `Optional` of them) works like a struct: body, return annotation and `Form()`, with the same 422. Defaults are filled in. It is checked by msgspec if installed, else pydantic; with neither, the route raises a `TypeError` telling you to `pip install msgspec`. A `NamedTuple` or attrs class is not supported. Before Python 3.12, pydantic wants `TypedDict` from `typing_extensions`.

```
from dataclasses import dataclass
from typing import TypedDict

from korvex import App
from korvex.testing import TestClient

app = App()


@dataclass
class NewTask:
    title: str
    done: bool = False


class Summary(TypedDict):
    title: str
    done: bool


@app.post("/tasks", status=201)
def add(task: NewTask) -> Summary:
    return {"title": task.title, "done": task.done}


print(TestClient(app).post("/tasks", json={"title": "milk"}).json())
```

## Headers, cookies and forms

```
from korvex import App, Cookie, Form, Header, Query, UploadedFile

app = App()


@app.post("/photos")
async def upload(
    title: str = Form(),  # a form field, urlencoded or multipart
    photo: UploadedFile = Form(),  # a file from a multipart form
    x_client: str = Header("web"),  # the X-Client header, "web" if missing
    theme: str = Cookie("light"),
    page_size: int = Query(20, alias="page-size"),
): ...
```

- `x_client` reads `X-Client`: underscores become hyphens, and case doesn't matter. `alias=` gives the exact name. `list[str]` gets every value of a repeated header.
- An empty form field counts as missing, except for a `str`.
- A whole model can come from a form, field by field: `signup: Signup = Form()`. A checkbox's `on` is true, and text converts to numbers and dates as the model says.
- All four also work in `Annotated[...]` (from `typing`): `theme: Annotated[str, Cookie()] = "light"`.

## What the handler returns

- A model, with a return annotation (`-> User`): sent as JSON, with the route's `status` (200 by default).
- `None`, annotated `-> None`: an empty response, for `status=204`.
- A `Response`: sent as it is, with its own status.

## Errors

Invalid input fails [validation](https://3d3n-pyc.github.io/korvex/glossary/#validation) and gets a 422 listing the problems:

```
{"detail": [{"loc": ["body", "age"], "msg": "Expected `int`, got `str`", "type": "validation"}]}
```

`loc` says where: `["query", "page"]`, `["header", "x-client"]`, `["form", "title"]`. pydantic lists every problem; msgspec stops at the first. To answer differently, handle the 422; the exception is a `RequestValidationError`, with the list in `errors`:

```
from korvex import App, RequestValidationError, Response

app = App()


@app.exception_handler(422)
def invalid(request, exc: RequestValidationError):
    return Response.json({"problems": [e["loc"] for e in exc.errors]}, status=422)
```

Annotations are read once, when the route is declared.

Next: [Requests and responses](https://3d3n-pyc.github.io/korvex/requests-and-responses/index.md).

# Requests and responses

Use this when a [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler) needs the request's details, or must build a specific response.

## What a handler gets

A handler's parameters say what it needs. Most handlers never touch the request itself:

```
import msgspec

from korvex import App

app = App()


class Task(msgspec.Struct):
    id: int
    title: str


@app.get("/tasks/{id:int}")
def get_task(id: int, verbose: bool = False) -> Task:  # path, then query string
    return Task(id=id, title="write docs")
```

When it does need it, it asks for it: a parameter named `request`, or annotated `Request`.

```
from korvex import App, Request, Response

app = App()


@app.post("/echo")
async def echo(request: Request):
    return Response(await request.body())
```

[Validation](https://3d3n-pyc.github.io/korvex/validation/index.md) lists everything a parameter can come from.

## The request

| Attribute                  |                                                                                                                               |
| -------------------------- | ----------------------------------------------------------------------------------------------------------------------------- |
| `method`, `path`, `scheme` | `"GET"`, `"/tasks/7"`, `"https"`                                                                                              |
| `headers`                  | `headers.get("accept")` the first value, `headers.get_all("accept")` every one; names in any case                             |
| `query`                    | the query string: `query.get("page")`, `query.get_all("tag")`                                                                 |
| `cookies`                  | a dict                                                                                                                        |
| `path_params`              | a dict                                                                                                                        |
| `root_path`                | the prefix the app is [mounted](https://3d3n-pyc.github.io/korvex/routes/#mounting-another-app) or served under, `""` if none |
| `client`                   | `(host, port)` of the client (the real one behind a trusted proxy), or `None`                                                 |
| `app`                      | the `App`                                                                                                                     |

And three methods to read the body:

- `await request.body()`: the bytes.
- `await request.json()`: parsed JSON (a 400 if it isn't).
- `await request.form()`: a form, urlencoded or multipart, read like the query (`form.get("title")`). Files are `UploadedFile`s, with `filename`, `content_type` and `data`. A body that isn't a form gets a 415; text that can't be decoded, a 400.

A [`WebSocket`](https://3d3n-pyc.github.io/korvex/glossary/#websocket) has the same attributes but no body: both are a `Connection`.

## Responses

```
from korvex import Response


def chunks():  # your generator
    yield "part one, "
    yield b"part two"


Response("raw bytes or text", status=200, headers={"X-Total": "3"})
Response.text("hello")
Response.json({"id": 1})
Response.redirect("/login", status=303)
Response.stream(chunks)  # a generator, sync or async, of bytes or str
Response.file("report.pdf", filename="report.pdf")  # your file; read in chunks; `filename` makes it a download
```

A handler can also return a model (`-> Task`, sent as JSON) or `None`: see [Validation](https://3d3n-pyc.github.io/korvex/validation/#what-the-handler-returns).

`headers=` takes a dict, or `(name, value)` pairs to repeat a name. A `Content-Type` given there wins over the one `text()` or `json()` sets.

`response.headers` is a `Headers`, the same as `request.headers`: names in any case, and a name can repeat (several `Set-Cookie`).

|                                                 |                                                     |
| ----------------------------------------------- | --------------------------------------------------- |
| `get(name)`, `headers[name]`                    | the first value (`[]` raises `KeyError` if missing) |
| `get_all(name)`                                 | every value                                         |
| `set(name, value)`, `headers[name] = value`     | replaces every value                                |
| `add(name, value)`                              | adds one more                                       |
| `remove(name)`                                  | removes them all                                    |
| `name in headers`, `for name, value in headers` | test, and go through the pairs                      |

### Cookies

```
from korvex import Response

token = "abc123"  # your value
response = Response.redirect("/")
response.set_cookie("session", token, max_age=3600, httponly=True, secure=True)
response.delete_cookie("theme")
```

`samesite` is `"lax"` unless you say otherwise. Without `max_age`, the cookie lasts until the browser closes.

### Streams and files

`Response.stream()` sends each chunk as it comes, to a slow client too: once 1 MiB waits for it, the generator waits. `Response.file()` reads the file in chunks, without blocking the server, and sets its type and length.

```
from korvex import App, Request, Response

app = App()


@app.get("/video")
def video(request: Request):
    return Response.file("movie.mp4", request=request)
```

With `request=`, the file gets an [`ETag`](https://3d3n-pyc.github.io/korvex/glossary/#etag), a `Last-Modified` and `Accept-Ranges`, and the response follows the request: 304 if the browser has the current file, 206 for a [`Range`](https://3d3n-pyc.github.io/korvex/glossary/#range-request) (resumable downloads, video seeking), 416 for a range past the end. Without it, the whole file is sent. A `compress()` [hook](https://3d3n-pyc.github.io/korvex/glossary/#hook) leaves 206 and 304 alone and makes the `ETag` weak when it compresses a file. [`app.static()`](https://3d3n-pyc.github.io/korvex/routes/#static-files) works this way.

### Server-Sent Events

[Server-Sent Events](https://3d3n-pyc.github.io/korvex/glossary/#server-sent-events) push a stream of events to a browser.

```
import asyncio

from korvex import App, Header, Response, ServerSentEvent

app = App()


@app.get("/feed")
async def feed(last_event_id: int = Header(0)):
    async def events():
        try:
            for n in range(last_event_id + 1, 1000):
                yield ServerSentEvent({"n": n}, event="tick", id=str(n))
                await asyncio.sleep(1)
        finally:
            ...  # runs when the client leaves
    return Response.sse(events(), ping=15.0)
```

A `str` is sent as is, anything else as JSON. In an async generator, a `: ping` comment goes out after `ping` seconds of silence (15 by default, `ping=0` turns it off); a sync generator gets none. The response sets `Content-Type: text/event-stream`, `Cache-Control: no-cache` and `X-Accel-Buffering: no` (for nginx), and `compress()` leaves it alone. When the client leaves, the generator is closed and its `finally` runs. A browser that reconnects sends the last `id` in `Last-Event-ID`; read it with `Header()`. A newline in `event` or `id` raises `ValueError`.

The event is encoded when you create it, so a bad value fails there: `bytes` data raises `TypeError`; a lone surrogate or `NaN` raises `ValueError`.

On the built-in server, a client that left is noticed at the next write. Keep `ping` on (15 s by default): the generator is closed within `ping` seconds. With `ping=0` or a sync generator, only when the next event is sent. Under ASGI servers it is noticed right away.

## Background tasks

Ask for `BackgroundTasks` to run [background tasks](https://3d3n-pyc.github.io/korvex/glossary/#background-task): work that goes on after the response is sent.

```
import msgspec

from korvex import App, BackgroundTasks

app = App()


class NewUser(msgspec.Struct):
    email: str
    name: str


class User(msgspec.Struct):
    email: str
    name: str


def create_user(body: NewUser) -> User: ...  # your code


def send_welcome_email(email: str, name: str): ...  # your code, blocking smtplib


@app.post("/signup", status=201)
def signup(body: NewUser, tasks: BackgroundTasks) -> User:
    user = create_user(body)
    tasks.add(send_welcome_email, user.email, name=user.name)
    return user
```

A handler and its dependencies share one object. Tasks run in the order added, after the response, the after-hooks, a stream's end and dependency teardown; also if the client left.

- A coroutine function is awaited; a plain function runs in a thread.
- An error is logged on the `"korvex"` logger, the next task still runs, the client never sees it. `TestClient` raises it (`raise_server_errors=False` gives just the response) and waits for the tasks, so a test can check their effect right away.
- Tasks run only if the handler returned, not after an `HTTPException` or a 500.
- There is no limit on how many run at once: for heavy or many tasks, use a job queue.
- A task should not block forever: shutdown cuts it off.
- A WebSocket route cannot ask for it: `TypeError` when declared.
- On shutdown the built-in server waits for running tasks up to `graceful_timeout`, then cuts them off with a warning.

Tasks run in the worker's process: for work that must survive a restart or needs retries, use a real queue.

Next: [Dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md).

# Templates

Use this when your app answers with HTML pages. Korvex renders them with its own template engine. No extra package.

```
from korvex import App

app = App()
app.static("/static", "static")  # the folder static/ holds css/app.css
views = app.templates("templates")  # a folder of .html files


def load_post(id: int) -> dict:  # your code: read the post from your database
    return {"id": id, "title": "Hello", "draft": False, "tags": ["a", "b", "c"]}


@app.get("/posts/{id}")
def show(id: int):
    post = load_post(id)
    return views.response("post.html", {"post": post})
```

`templates/base.html`:

```
<!doctype html>
<title>{% block title %}Blog{% endblock %}</title>
<link rel="stylesheet" href="{{ static_url('css/app.css') }}">
<main>{% block content %}{% endblock %}</main>
{% include "footer.html" %}
```

`templates/post.html`:

```
{% extends "base.html" %}
{% block title %}{{ post.title }}{% endblock %}
{% block content %}
  <h1>{{ post.title|upper }}</h1>
  {% if post.draft %}
    <p>Draft</p>
  {% elif post.tags|length > 2 %}
    <p>Popular</p>
  {% endif %}
  <ul>
    {% for tag in post.tags %}
      <li>{{ loop.index }}/{{ loop.length }} {{ tag }}</li>
    {% else %}
      <li>No tags</li>
    {% endfor %}
  </ul>
{% endblock %}
```

`templates/footer.html` is any HTML, like `<footer>Acme</footer>`. `views.response(name, context)` renders the template with a dict of variables and returns an HTML response. It takes `status=` and `headers=` too. Templates are compiled the first time they are used, then cached.

## Syntax

- `{{ expression }}` prints a value. `{# comment #}` prints nothing.
- `{% if %}`, `{% elif %}`, `{% else %}`, `{% endif %}`.
- `{% for x in items %}`, with an optional `{% else %}` for an empty list. `for k, v in d.items()` unpacks. Inside a loop: `loop.index` (from 1), `loop.index0`, `loop.first`, `loop.last`, `loop.length`.
- `{% set total = price * 3 %}`: visible from there to the end of the template, also after the `for` or `if` it was set in.
- `{% extends "base.html" %}` (first tag) and `{% block name %}...{% endblock %}`. In a child template, anything outside the blocks is ignored.
- `{% include "nav.html" %}`. The name is a string in quotes.
- `{%- ... -%}` and `{{- ... -}}` remove the whitespace before or after.

An expression can use names, `a.b`, `a[k]`, `a[1:3]`, strings, numbers, `True`, `False`, `None`, lists, dicts, comparisons, `and`, `or`, `not`, `in`, `+ - * / // %`, `x if cond else y`, calls, and filters (`value|filter` or `value|filter(arg)`). `a.b` reads the key `b` of a dict, the attribute `b` of anything else.

Not supported: macros, `is` tests other than `is None`, comprehensions, lambdas. A syntax error names the file and the line.

## Filters

| Filter                   | Does                                                                                     |
| ------------------------ | ---------------------------------------------------------------------------------------- |
| `safe`                   | Do not escape: the value is trusted HTML.                                                |
| `escape`, `e`            | Escape the value.                                                                        |
| `upper`, `lower`         | Change the case.                                                                         |
| `default(x)`             | `x` when the value is `None` or undefined.                                               |
| `length`                 | Number of items or characters.                                                           |
| `join(sep)`              | Join a list, escaping each item.                                                         |
| `truncate(n, end="...")` | Cut to `n` characters at a word boundary.                                                |
| `tojson`                 | The value as JSON, safe in a `<script>` and in a single-quoted attribute (\`data-x='{{ x |
| `urlencode`              | Percent-encode a text, or a dict as a query string.                                      |

Add your own, or a value every template can read:

```
views.filters["money"] = lambda cents: f"{cents / 100:.2f} EUR"
views.globals["site_name"] = "Acme"
```

`{{ price|money }}` and `{{ site_name }}` now work. `url_for` and `static_url` are already there:

```
<a href="{{ url_for('show', id=post.id) }}">{{ post.title }}</a>
```

## Escaping

- Every printed value is HTML-escaped, unless it is a `Markup` (from `korvex`) or goes through `|safe`. `None` prints as an empty string.
- In `{% if %}` and `{% elif %}`, a missing name, key or attribute counts as `None` (false), and an ordering comparison such as `post.n > 3` is false. Elsewhere it is an error, with the file and line: `post.html, line 5: 'post.draft' is undefined`, unless you use `{{ post.draft|default(False) }}`.
- To give data to JavaScript, use `tojson`:

```
<script>const items = {{ items|tojson }};</script>
```

- Escaping does not make a URL safe: `href="javascript:..."` stays a script. Only put URLs you trust in `href` and `src`, and use `|urlencode` for a value in a query string: `<a href="/search?q={{ q|urlencode }}">`.

## Trust

Templates are code, written by you. Names and attributes that start with an underscore are refused, but this is not a sandbox. Do not render templates written by your users.

`format` and `format_map` are refused too: they can reach attributes from inside the string; use a filter instead.

## Reload

```
views = app.templates("templates", reload=True)
```

With `reload=True`, a template is compiled again when it, or a template it extends or includes, changes on disk. Use it in development. It is off by default: nothing checks the files on each render.

## Static files with a hash

`static_url()` gives the URL of a file served by `app.static()`, with a hash of its content:

```
<link rel="stylesheet" href="{{ static_url('css/app.css') }}">
<!-- /static/css/app.3f2a1c9b.css -->
```

- The server answers that URL with `Cache-Control: public, max-age=31536000, immutable`, because the URL changes when the file does. Browsers keep it for a year.
- The plain URL (`/static/css/app.css`) still works, without that header.
- With several `app.static()` mounts, pick one with `static_url('logo.svg', name="assets")`.
- In a [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler), call `app.static_url("css/app.css")`. It raises `FileNotFoundError` for a missing file and `LookupError` for an unknown mount.

## Testing

```
from korvex.testing import TestClient

with TestClient(app) as client:
    assert "<h1>HELLO</h1>" in client.get("/posts/1").text

html = views.render("post.html", {"post": load_post(1)})  # the page as a str
```

## Speed

4 to 6 times faster than Jinja2 in `benches/bench_templates.py`. See [Performance](https://3d3n-pyc.github.io/korvex/performance/index.md).

Next: [Sessions and CSRF](https://3d3n-pyc.github.io/korvex/sessions/index.md).

# Sessions and CSRF

Use this when visitors sign in, or you need to remember something between their requests. `sessions(app, ...)` keeps a small dict per browser in a signed cookie ([session](https://3d3n-pyc.github.io/korvex/glossary/#session)). `csrf(app)` refuses cross-site requests that change things ([CSRF](https://3d3n-pyc.github.io/korvex/glossary/#csrf)). No store, no extra package, no tokens in your forms.

```
import os
from typing import Optional

from korvex import App, Depends, Form, HTTPException, Response
from korvex.middleware import csrf, current_session, flash, sessions

app = App()
sessions(app, secret_key=os.environ["SECRET_KEY"])  # see below to make one
csrf(app)


class User:  # your code
    def __init__(self, id: int, name: str):
        self.id, self.name = id, name


def check_password(username: str, password: str) -> Optional[User]:
    # your code: returns the user, or None
    return User(1, username) if password == "pw" else None


@app.post("/login")
def login(username: str = Form(), password: str = Form()):
    user = check_password(username, password)
    if user is None:
        flash("Wrong name or password", "error")
        return Response.redirect("/login", status=303)
    session = current_session()
    session.clear()  # never keep what an anonymous visitor had (session fixation)
    session["user_id"] = user.id
    flash(f"Welcome, {user.name}", "success")
    return Response.redirect("/", status=303)


def current_user() -> int:
    user_id = current_session().get("user_id")
    if user_id is None:
        raise HTTPException(401, "log in first")
    return user_id


@app.get("/account")
def account(user_id: int = Depends(current_user)):
    return Response.json({"user_id": user_id})


@app.post("/logout")
def logout():
    current_session().clear()
    return Response.redirect("/", status=303)
```

Make a key once and keep it out of the code:

```
python -c "import secrets; print(secrets.token_urlsafe(32))"
```

Put the output in `SECRET_KEY`. A key must be at least 32 bytes, or `sessions()` raises `ValueError`. To rotate, pass a list with the new key first: it signs, and every key in the list is still accepted when checking. Drop the old key once its sessions have expired.

```
sessions(app, secret_key=[os.environ["NEW_KEY"], os.environ["OLD_KEY"]])  # in place of the one above
```

## What a session is

`current_session()` returns a `Session`: a dict stored in a signed cookie.

- The browser can read it, as it is not encrypted. Put no secret in it. It cannot change it: a cookie that was edited, or signed with another key, gives an empty session.
- Values must be JSON: strings, numbers, booleans, `None`, lists, dicts.
- The signed cookie may be 4096 bytes. A bigger session raises `ValueError`. Keep ids in it, not data.
- The cookie is sent only when the session was read and changed. Setting, deleting or clearing a key counts. A change inside a nested value does not, so set `session.modified = True` after it.
- `max_age` counts from the last change, not from the last visit. A session that is only read is not renewed.
- Clearing a session that came with a cookie deletes the cookie.

```
session = current_session()
cart = session.setdefault("cart", [])
cart.append(3)
session.modified = True
```

| Option       | Default     | What it does                                             |
| ------------ | ----------- | -------------------------------------------------------- |
| `secret_key` | (required)  | A key, or a list of keys (new first). 32 bytes at least. |
| `max_age`    | 14 days     | Seconds a session stays valid.                           |
| `cookie`     | `"session"` | The cookie's name.                                       |
| `secure`     | `True`      | Send the cookie over HTTPS only.                         |
| `samesite`   | `"lax"`     | The cookie's `SameSite` attribute.                       |
| `path`       | `"/"`       | The cookie's path.                                       |
| `domain`     | none        | The cookie's domain. Unset: the exact host only.         |

The cookie is always `HttpOnly`: scripts on the page cannot read it. With `secure=True`, browsers do not send it over plain `http://`, except to `localhost`. Use `secure=False` only for a local setup that is not `localhost`.

`sessions()` can be called once per app. `current_session()` raises `RuntimeError` outside a request, or when `sessions()` was not called.

## Limits to know

- **A session cannot be revoked.** The cookie is the whole session. A copy of it stays valid until it expires, even after the user logged out. To be able to cancel sessions, keep something on the server and check it: for example a version number per user, stored in the session at login and compared in `current_user`. Raise the number to log that user out everywhere.
- **In a WebSocket handler the session is read-only.** `current_session()` reads the cookie of the handshake. A change is lost, as the cookie can not be sent after the handshake.
- **Register order does not matter.** `sessions()` always runs first among the before hooks, so a `csrf()` or `rate_limit()` refusal handler can read the session too.
- **A streamed response cannot change the session.** Its headers are sent before the body: a stream can read the session, but a change made while it is being sent is not saved.

## Flash messages

`flash(message, category="info")` keeps a message in the session. The next page shows it once.

```
flash("Saved", "success")
```

In a template made with `app.templates()`, `get_flashed_messages()` is available. It returns `(category, message)` pairs and forgets them:

```
{% for category, message in get_flashed_messages() %}
  <p class="flash {{ category }}">{{ message }}</p>
{% endfor %}
```

In Python, import `get_flashed_messages` from `korvex.middleware`. Both `flash()` and `get_flashed_messages()` change the session, so they need `sessions(app, ...)`.

## CSRF: `csrf(app)`

A cross-site request forgery is a page on another site that makes the visitor's browser send a `POST` to yours, with your cookies. `csrf(app)` refuses it with a 403. There is nothing to add to your forms or `fetch` calls.

It uses what the browser sends, in this order:

1. `Sec-Fetch-Site` is present: `same-origin` and `none` (the user typed the address) pass. Anything else, `same-site` included, is refused, unless the `Origin` is in `trusted_origins`.
1. Else `Origin` is present: it passes if it equals the request's `Host` (scheme aside), or is in `trusted_origins`.
1. Neither header: the client is not a browser (curl, a script, a mobile app), so it passes. A browser always sends one of them on a request that changes something.

`GET`, `HEAD` and `OPTIONS` always pass. WebSocket handshakes are checked too, as a page on another site can open one with the visitor's cookies.

There is no token because every current browser sends these headers. A very old browser sends neither, and is let through.

| Option            | What it does                                                                                                                 |
| ----------------- | ---------------------------------------------------------------------------------------------------------------------------- |
| `trusted_origins` | Other origins allowed, like `["https://app.example.com"]` for a separate front-end. Each is `scheme://host[:port]`, no path. |
| `exempt`          | Paths not checked. An exact path (`"/hook"`) or, ending in `/`, a prefix (`"/webhooks/"`).                                   |

```
csrf(
    app,
    trusted_origins=["https://app.example.com"],
    exempt=["/webhooks/"],  # these check their own signature
)
```

A bad entry raises `ValueError` when `csrf()` is called.

Behind a reverse proxy, `Origin` is compared with the `Host` header as the app receives it. If the proxy replaces `Host` (with `127.0.0.1:8000`, for example), every form post is refused. Keep the public `Host` (nginx: `proxy_set_header Host $host;`; Caddy does by default), let the built-in server take the proxy's `X-Forwarded-Host` (`--forwarded-host`), or list the public origin in `trusted_origins`.

`csrf()` and `cors()` are separate. A cross-origin front-end allowed by `cors()` must also be listed in `csrf(trusted_origins=...)`. Without a `Sec-Fetch-Site` header, the `Origin` check compares hosts, not schemes (as Go does).

## Testing

`TestClient` keeps cookies, so a session lasts across the requests of one client. It also sends `Secure` cookies back to `http://testserver` and to `localhost`, as browsers do, so the default `secure=True` works in tests.

```
from korvex.testing import TestClient

client = TestClient(app)  # the app above
client.post("/login", form={"username": "ada", "password": "pw"})
assert client.get("/account").status == 200
```

To test CSRF, send the header a browser would send from another site:

```
r = client.post("/logout", headers={"Sec-Fetch-Site": "cross-site"})
assert r.status == 403
```

Next: [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md).

# Dependencies

Use this when several routes need the same code run first, and its result.

A [dependency](https://3d3n-pyc.github.io/korvex/glossary/#dependency) is a function whose result a handler receives. Put code shared by routes there: who is signed in, a database connection, a permission check.

```
from typing import Annotated, Optional

import msgspec

from korvex import App, Cookie, Depends, HTTPException

app = App()


class User(msgspec.Struct):
    id: int
    name: str


class Users:  # your storage
    async def by_session(self, session): ...


users = Users()


async def current_user(session: Optional[str] = Cookie(None)) -> User:
    user = await users.by_session(session)
    if user is None:
        raise HTTPException(401, "sign in first")  # the request stops here
    return user


@app.get("/me")
def me(user: User = Depends(current_user)) -> User:
    return user


CurrentUser = Annotated[User, Depends(current_user)]  # one name for every use
```

- A dependency takes parameters like a handler: query, headers, cookies, body, path, the request, and other dependencies. It can be sync or async.
- It runs once per request, however many parameters ask for it. `Depends(func, cache=False)` runs it each time.
- `Annotated` names it once for all uses (`CurrentUser` above), then `def me(user: CurrentUser)`.

## Guarding routes

To check without using the result, list it in `dependencies=`, on a route, a group or a WebSocket:

```
from korvex import App, Depends


async def current_user(): ...  # as above


app = App()
api = app.group("/api", dependencies=[Depends(current_user)])  # every route of /api
```

A group's dependencies run before its routes' own.

## Setup and teardown

A dependency that `yield`s is torn down after the handler:

```
from typing import Annotated

import msgspec

from korvex import App, Depends

app = App()


class Task(msgspec.Struct):
    id: int


class User(msgspec.Struct):
    id: int


class Pool:  # your connection pool
    async def acquire(self): ...

    async def release(self, connection): ...


pool = Pool()


async def current_user() -> User: ...  # as above


CurrentUser = Annotated[User, Depends(current_user)]  # as above


async def database():
    connection = await pool.acquire()
    try:
        yield connection  # the handler runs here
    finally:
        await pool.release(connection)


@app.get("/tasks")
async def tasks(user: CurrentUser, db=Depends(database)) -> list[Task]:
    return await db.tasks_of(user.id)
```

- For a streamed response, the teardown waits until the stream is sent.
- An error from the handler is raised at the `yield`, so `except` and `finally` work there. The error still gets its usual response.

## In tests

```
from korvex import App

app = App()


def current_user(): ...  # as above


class User:  # as above
    def __init__(self, id, name): ...


app.dependency_overrides[current_user] = lambda: User(id=1, name="test")
```

replaces a dependency everywhere. See [Testing](https://3d3n-pyc.github.io/korvex/testing/index.md).

## Errors

Validation errors from dependencies and from the handler come together in one 422. A dependency can also `raise Abort(response)` to send that exact response: a redirect to the sign-in page, say. See [Errors and hooks](https://3d3n-pyc.github.io/korvex/errors-and-hooks/index.md).

Next: [Errors and hooks](https://3d3n-pyc.github.io/korvex/errors-and-hooks/index.md).

# Errors and hooks

Use this when you need to answer an error your way, run code on every request, or open and close something (a database pool) with the app.

## Errors

```
from korvex import App, HTTPException

app = App()


@app.get("/tasks/{id}")
def task(id: int):
    raise HTTPException(404, "no such task")  # {"detail": "no such task"}, a 404
```

- Raise it in a [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler), a [dependency](https://3d3n-pyc.github.io/korvex/glossary/#dependency) or a before [hook](https://3d3n-pyc.github.io/korvex/glossary/#hook). Without a message, the standard one is used (`"Not Found"`). `headers=` adds headers: `{"WWW-Authenticate": "Bearer"}`.
- Every error Korvex answers is JSON with a `detail`: these, routing's 404 and 405, validation's 422.
- Any other exception gives a 500, `{"detail": "Internal Server Error"}`. The error itself is logged to the `korvex` logger, never sent.

`Abort` sends an exact response instead, past the exception handlers:

```
from typing import Optional

from korvex import Abort, Cookie, Depends, Response


def signed_in(session: Optional[str] = Cookie(None)) -> None:
    if session is None:
        raise Abort(Response.redirect("/login"))


@app.get("/private", dependencies=[Depends(signed_in)])
def private():
    return Response.text("secret")
```

After hooks still see that response. In a WebSocket's dependency, `Abort` refuses the connection (a 403).

### Answering errors your way

```
@app.exception_handler(404)  # a status
def not_found(request, exc):
    return Response.text("Nothing here", status=404)


@app.exception_handler(ValueError)  # an exception class, and its subclasses
def bad_value(request, exc: ValueError):
    return Response.json({"detail": str(exc)}, status=400)


@app.get("/boom")
def boom():
    raise ValueError("bad value")  # answered by bad_value: a 400
```

## Hooks around every request

```
from korvex import Response


@app.before_request
def block_bots(connection):
    if "bot" in (connection.headers.get("user-agent") or ""):
        return Response.text("no bots", status=403)  # answers right away
    # returning None carries on


@app.after_request
def add_header(request, response):
    response.headers.set("X-Served-By", "korvex")  # change it, or return another
```

- Before hooks run before routing, so for unknown paths too (and with `path_params` still empty), in the order declared. They also run for WebSocket handshakes: they get a `Connection`, and `isinstance(connection, WebSocket)` tells a handshake apart. Answering refuses the connection.
- After hooks get every response, errors included, and run in reverse order, like layers.
- Both can be sync or async. Ready-made ones are in [korvex.middleware](https://3d3n-pyc.github.io/korvex/middleware/index.md).

## Startup and shutdown

```
import asyncpg  # pip install asyncpg

DATABASE_URL = "postgresql://localhost/mydb"  # your code


@app.on_startup
async def connect():
    app.local.pool = await asyncpg.create_pool(DATABASE_URL)


@app.on_shutdown
async def disconnect():
    await app.local.pool.close()
```

- Startup hooks run before the first request; if one raises, the server doesn't start. Shutdown hooks run when it stops, last declared first.
- `app.local` keeps attributes per event loop: with the built-in server's threads, each [event loop](https://3d3n-pyc.github.io/korvex/glossary/#event-loop) gets its own pool, as asyncio needs. Use it instead of module globals for what a startup hook opens, and read it from handlers and dependencies:

```
async def db():  # a dependency, used with Depends(db)
    async with app.local.pool.acquire() as connection:
        yield connection
```

Next: [WebSockets](https://3d3n-pyc.github.io/korvex/websockets/index.md).

# WebSockets

Use this when the browser and the server need to send each other messages at any time, such as a chat. A [WebSocket](https://3d3n-pyc.github.io/korvex/glossary/#websocket) is one open connection for that.

```
from typing import Annotated

import msgspec

from korvex import App, Depends, HTTPException, WebSocket

app = App()


class User(msgspec.Struct):
    name: str


def current_user(token: str = "") -> User:  # your code: who is connecting
    if not token:
        raise HTTPException(401, "token needed")  # refuses the connection
    return User(name=token)


CurrentUser = Annotated[User, Depends(current_user)]


@app.websocket("/chat/{room}")
async def chat(ws: WebSocket, room: str, user: CurrentUser):
    await ws.accept()
    async for message in ws:  # text or bytes, until the client leaves
        await ws.send_text(f"{user.name} in {room}: {message}")
```

- The [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler) gets the `WebSocket` by naming a parameter `ws` or `websocket`, or annotating it.
- Before `accept()`, the handler can look at the handshake: `headers`, `cookies`, `query`, `path_params`, `client`. `close()` before accepting refuses the connection (a 403).
- `receive()` gives the next message, `receive_text()`, `receive_bytes()` and `receive_json()` one of that kind; they raise `WebSocketDisconnect` once the client is gone. `send_text()`, `send_bytes()`, `send_json()` and `close(code)` go the other way; `close()` once the client is gone does nothing.
- Parameters and [dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md) work as for HTTP, without a body. A dependency raising `HTTPException` refuses the connection. A dependency both kinds of route share takes a `Connection`, not a `Request`.
- Groups and hosts declare WebSockets too: `api.websocket(...)`.
- [Before hooks](https://3d3n-pyc.github.io/korvex/errors-and-hooks/#hooks-around-every-request) run on the handshake: check its `Origin` there, which browsers don't check for WebSockets. After hooks don't run: there's no response.

They work the same under any [ASGI](https://3d3n-pyc.github.io/korvex/glossary/#asgi) server and in the [built-in server](https://3d3n-pyc.github.io/korvex/server/index.md), which also pings idle connections and makes a slow client wait instead of piling messages in memory.

Next: [OpenAPI and security](https://3d3n-pyc.github.io/korvex/openapi/index.md).

# OpenAPI and security

Use this when you want API documentation (Swagger UI) built from your [routes](https://3d3n-pyc.github.io/korvex/glossary/#route), and a way to sign in from it.

```
from korvex import App

app = App()
app.docs(title="Tasks", version="1.0")  # /openapi.json and Swagger UI at /docs
```

The document is built from the routes: paths, typed parameters, bodies, returned [models](https://3d3n-pyc.github.io/korvex/glossary/#model), and each [handler](https://3d3n-pyc.github.io/korvex/glossary/#handler)'s docstring (its first line is the summary). `app.openapi()` gives it as a dict. Swagger UI comes from a CDN; `docs_path=None` leaves it out.

## What else to say about a route

```
from uuid import UUID

import msgspec

from korvex import App, Response


class Task(msgspec.Struct):  # your models
    id: UUID
    title: str


class NewTask(msgspec.Struct):
    title: str


class Conflict(msgspec.Struct):
    detail: str


app = App()
api = app.group("/api", tags=["tasks"], responses={401: "Not signed in"})


@api.post("/tasks", status=201, responses={409: ("Title taken", Conflict)})
def create_task(body: NewTask) -> Task:
    return Task(id=UUID(int=1), title=body.title)


@api.delete("/tasks/{id:uuid}", status=204, responses={404: "No such task"})
def delete_task(id: UUID) -> None: ...


@app.get("/v1/tasks", deprecated=True)
def old_tasks() -> list[Task]: ...


@app.get("/health", include_in_schema=False)
def health():
    return Response.text("ok")
```

- `status`: the success, and the status of what the handler returns (a model, or `None` for an empty response). A returned `Response` keeps its own.
- `responses`: other answers, by status: a description, a model, or both (`(description, model)`). `"4XX"` and `"default"` work too.
- `tags`, `deprecated`, `include_in_schema`: as they say.
- On a group or a host, they apply to its routes: tags add up, and a route's responses replace the group's for the same status.

## Signing in from Swagger UI

`korvex.security` reads a credential from the request, refuses the request without one (401), and tells OpenAPI, so Swagger UI's "Authorize" button sends it:

```
from typing import Optional

import msgspec

from korvex import App, Depends, HTTPException
from korvex.security import Bearer


class User(msgspec.Struct):
    name: str


class Users:  # your code: look a token up in your database
    async def by_token(self, token: str) -> Optional[User]:
        return User("ada") if token == "t0ken" else None


users = Users()
app = App()
bearer = Bearer()  # Authorization: Bearer <token>


async def current_user(token: str = Depends(bearer)) -> User:
    user = await users.by_token(token)
    if user is None:
        raise HTTPException(401, "invalid token", headers={"WWW-Authenticate": "Bearer"})
    return user


@app.get("/me")
def me(user: User = Depends(current_user)) -> User:
    return user
```

- `Bearer(format="JWT")`, `APIKey(header="X-API-Key")` (or `query=`, `cookie=`) and `OAuth2Password(token_url="/token", scopes={...})` (with `refresh_url=` if tokens can be refreshed). Prefer a header for an API key: one in the [query string](https://3d3n-pyc.github.io/korvex/glossary/#query-string) ends up in access logs.
- Each takes `name=` (its name in OpenAPI) and `description=` (shown in Swagger UI).
- A scheme only reads the credential; the [dependency](https://3d3n-pyc.github.io/korvex/glossary/#dependency) that takes it checks it, like `current_user` above.
- `required=False` gives `None` instead of a 401.
- A route lists every scheme it depends on, directly, through another dependency or through its group.

### Your own scheme

Subclass `SecurityScheme`: `credential()` finds it in the request, `openapi()` describes it, and `challenge` goes in the 401's `WWW-Authenticate`. HTTP Basic, for example:

```
from korvex import App, Depends, Response
from korvex.security import SecurityScheme


class Basic(SecurityScheme):
    challenge = "Basic"

    def __init__(self):
        super().__init__(name="basic", description=None, required=True)

    def credential(self, request):
        scheme, _, value = request.headers.get("authorization", "").partition(" ")
        return value if scheme.lower() == "basic" else None

    def openapi(self):
        return {"type": "http", "scheme": "basic"}


app = App()


@app.get("/basic")
def basic(credential: str = Depends(Basic())):
    return Response.json({"credential": credential})  # base64 of user:password
```

Next: [Ready-made middleware](https://3d3n-pyc.github.io/korvex/middleware/index.md).

# Ready-made middleware

Use this when you want common protections (CORS, rate limiting, security headers...) without writing them. `korvex.middleware` has ready-made [middleware](https://3d3n-pyc.github.io/korvex/glossary/#middleware): each is a function that registers a [hook](https://3d3n-pyc.github.io/korvex/glossary/#hook) on the app:

```
from korvex import App
from korvex.middleware import compress, cors, csrf, rate_limit, request_id, security_headers, sessions

app = App()
compress(app)  # first: it compresses what the others produced
cors(app, allow_origins=["https://example.com"])
request_id(app)
rate_limit(app, requests=100, per=60)
security_headers(app)
csrf(app)  # refuses cross-site form posts
sessions(app, secret_key="x" * 32)  # signed cookie sessions; use a real secret key
```

Behind a [reverse proxy](https://3d3n-pyc.github.io/korvex/glossary/#reverse-proxy), let the proxy compress and add headers. Sessions, flash messages and CSRF protection are in the [Sessions and CSRF guide](https://3d3n-pyc.github.io/korvex/sessions/index.md).

## CORS: `cors(app, allow_origins=...)`

Lets browsers call the app from pages on other sites (origins): [CORS](https://3d3n-pyc.github.io/korvex/glossary/#cors).

| Option              | Default     | What it does                                                         |
| ------------------- | ----------- | -------------------------------------------------------------------- |
| `allow_origins`     | (required)  | The sites allowed, like `["https://example.com"]`, or `"*"` for any. |
| `allow_methods`     | all of them | The methods those pages may use.                                     |
| `allow_headers`     | none        | Request headers they may send beyond the basic ones, or `"*"`.       |
| `allow_credentials` | `False`     | Whether they may send cookies.                                       |
| `expose_headers`    | none        | Response headers their code may read.                                |
| `max_age`           | `600`       | Seconds the browser may remember a preflight answer.                 |

A browser's preflight request (`OPTIONS` with `Access-Control-Request-Method`) is answered before routing: 204 if allowed, 400 if not. Every other response to an allowed origin gets the CORS headers, errors included.

## Request IDs: `request_id(app)`

Gives each request an ID, sends it back in the response, and puts it in your logs.

| Option           | Default          | What it does                                                                                                                |
| ---------------- | ---------------- | --------------------------------------------------------------------------------------------------------------------------- |
| `header`         | `"X-Request-ID"` | The header the ID is read from and sent back in.                                                                            |
| `trust_incoming` | `True`           | Keep an ID the client or a proxy sent, when it's safe (letters, digits, `._:-`, up to 200). `False` always makes a new one. |

- `current_request_id()` returns it anywhere in your code, while the request runs.
- `RequestIdFilter` adds it to log records:

```
import logging
from korvex.middleware import RequestIdFilter

handler = logging.StreamHandler()
handler.addFilter(RequestIdFilter())
handler.setFormatter(logging.Formatter("%(request_id)s %(message)s"))
logging.getLogger("korvex").addHandler(handler)
```

## Rate limiting: `rate_limit(app, requests=100, per=60)`

Allows each client `requests` requests every `per` seconds; beyond, a 429 with `Retry-After`.

| Option            | Default              | What it does                                                                                                                              |
| ----------------- | -------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- |
| `requests`, `per` | (required)           | The allowance, like 100 per 60 seconds.                                                                                                   |
| `key`             | the client's address | Who the limit applies to: a function of the request, returning an API key, a user ID... or `None` to exempt the request (a health check). |
| `store`           | a `MemoryStore`      | Where the counts live: see below.                                                                                                         |
| `headers`         | `True`               | Send `RateLimit-Limit`, `RateLimit-Remaining` and `RateLimit-Reset`, so clients can slow down.                                            |

It counts WebSocket handshakes too. Requests can come in bursts of up to `requests`, then one every `per / requests` seconds.

- Hooks declared before it run first: call `cors()` before it, so browsers' preflight requests are answered without being counted.
- Behind a proxy, trust it with `--forwarded-allow-ips`, or every client counts as the proxy. Requests with no client address at all share one limit.

### Where the counts live

- `MemoryStore()`, the default: in each worker's memory. With 3 workers, a client gets up to 3 times the limit.
- `RedisStore("redis://localhost:6379/0")`: in Redis, shared by every worker and machine (`pip install korvex[redis]`). It takes a URL or a `redis.asyncio.Redis` you made yourself; `prefix=` sets what its keys start with. If Redis can't be reached, requests go through and the error is logged; `fail_open=False` refuses them instead.

### Your own store

A store (a `RateLimitStore`) is any object with a `hit()` method, sync or async, that decides for one request:

```
from korvex.middleware import Decision


class MyStore:
    async def hit(self, key, requests: int, per: float) -> Decision:
        # your code: count the hit, in Redis, a database...
        return Decision(allowed=True, remaining=99, retry_after=0, reset=0.6)


rate_limit(app, requests=100, per=60, store=MyStore())  # app: the one above
```

`Decision` has `allowed`, `remaining` (requests left), `retry_after` (seconds until the next one is allowed, 0 if allowed) and `reset` (seconds until the whole allowance is back). The store decides rather than just storing numbers, so a shared one can do it atomically.

## Compression: `compress(app)`

Compresses text-like responses for clients that accept it.

| Option          | Default                             | What it does                                                                                                                                                                                                                                                                       |
| --------------- | ----------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `encodings`     | `("zstd", "br", "gzip")`            | In order of preference, among those the client accepts. `zstd` needs Python 3.14, `br` the `brotli` package (`pip install korvex[brotli]`); those not available are skipped. On free-threaded Python, `br` is left out of the default: the `brotli` package turns the GIL back on. |
| `minimum_size`  | `500`                               | Smaller bodies are sent as they are.                                                                                                                                                                                                                                               |
| `levels`        | gzip 6, brotli 4, zstd 3            | Per encoding: `{"gzip": 9}`.                                                                                                                                                                                                                                                       |
| `content_types` | text, JSON, JavaScript, XML, SVG... | What's worth compressing: prefixes like `"text/"`, or suffixes like `"+json"`.                                                                                                                                                                                                     |

- Streams are compressed piece by piece, each piece sent at once, so a live stream isn't held back.
- A response that already has a `Content-Encoding`, says `Cache-Control: no-transform`, or has no body is left alone.
- Don't compress a page that holds a secret (a CSRF token) next to something the client sent: an attacker could guess the secret (the BREACH attack).

## Security headers: `security_headers(app)`

Adds the headers that tell browsers to protect your pages. A response that sets one itself keeps its own.

| Option                            | Default                             | What it does                                                                                                                      |
| --------------------------------- | ----------------------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
| `hsts`                            | one year                            | `Strict-Transport-Security`: browsers stick to HTTPS. Sent over HTTPS only.                                                       |
| `hsts_subdomains`, `hsts_preload` | `False`                             | Add those directives; only once every subdomain serves HTTPS.                                                                     |
| `content_security_policy`         | none                                | A CSP, like `"default-src 'self'"`. Off by default: the right one depends on your pages, and `/docs` loads Swagger UI from a CDN. |
| `frame_options`                   | `"DENY"`                            | `X-Frame-Options`: other sites can't show your pages in a frame. `"SAMEORIGIN"` allows your own.                                  |
| `content_type_options`            | `True`                              | `X-Content-Type-Options: nosniff`: browsers trust `Content-Type`.                                                                 |
| `referrer_policy`                 | `"strict-origin-when-cross-origin"` | `Referrer-Policy`.                                                                                                                |
| `cross_origin_opener_policy`      | `"same-origin"`                     | `Cross-Origin-Opener-Policy`.                                                                                                     |
| `permissions_policy`              | none                                | `Permissions-Policy`, like `"camera=(), microphone=()"`.                                                                          |

Each header option takes `None` to leave the header out.

All the details are in the [reference](https://3d3n-pyc.github.io/korvex/reference/middleware/index.md).

Next: [Testing](https://3d3n-pyc.github.io/korvex/testing/index.md).

# Testing

`korvex.testing` runs the app in your tests, without a server:

Use this when you want to check your app with automated tests (pytest, for example). Here is a tiny app and its test:

```
from korvex import App, Form, HTTPException, Response, WebSocket
from korvex.middleware import current_session, sessions
from korvex.testing import TestClient

app = App()
sessions(app, secret_key="x" * 32)  # a throwaway key, for the example


@app.post("/login")
def login(password: str = Form()):
    if password != "s3cret":
        raise HTTPException(401, "wrong password")
    current_session()["user"] = "ada"
    return Response.redirect("/", status=303)


@app.get("/api/tasks")
def tasks(done: bool = False):
    if "user" not in current_session():
        raise HTTPException(401, "log in first")
    return Response.json([])


@app.websocket("/live")
async def live(ws: WebSocket):
    await ws.accept()
    await ws.send_json({"event": "presence"})


def test_tasks():
    with TestClient(app) as client:
        assert client.post("/login", form={"password": "s3cret"}).status == 303
        response = client.get("/api/tasks", query={"done": "false"})
        assert response.json() == []  # the session cookie was sent back

        with client.websocket("/live") as ws:
            assert ws.receive_json()["event"] == "presence"
```

## With pytest

Put the client in a fixture, in `conftest.py`. The `with` block runs the app's startup and shutdown hooks around each test:

```
import pytest

from korvex.testing import TestClient

from myapp import app  # your code: where your App lives


@pytest.fixture
def client():
    with TestClient(app) as client:
        yield client
```

```
def test_tasks_need_login(client):
    assert client.get("/api/tasks").status == 401
```

## Requests

- `get()`, `post()`, `put()`, `patch()`, `delete()`, `head()` and `options()` take `query=`, `headers=`, `cookies=`, and a body: `json=`, `form=` (with `files={"doc": ("a.pdf", data, "application/pdf")}` for a multipart form), or raw `body=`.
- A response has `status`, `headers`, `body`, `text`, `json()`, and the `url` it came from.
- Cookies the app sets are kept in `client.cookies` and sent back.
- `follow_redirects=True` follows redirects, up to 20 (after a 303, or a 301/302 answering a POST, as a GET).
- A full URL sets the scheme and host: `client.get("https://acme.example.com/")`. Otherwise requests go to `base_url`, `http://testserver` by default.
- `headers=` and `cookies=` given to the client go with every request: `TestClient(app, headers={"Authorization": "Bearer t0ken"})`.

## What the client does for you

- In a `with` block, the app's startup hooks run first and its shutdown hooks last; without one, no hook runs. Every request runs on the same event loop.
- An error the app answers with a 500 is raised in the test, with its traceback. `raise_server_errors=False` gives you the 500 instead.
- A request or WebSocket message not received within `timeout` (5 seconds) raises `TimeoutError`; `timeout=None` waits forever.

## Streams and WebSockets

```
# `client` is a TestClient, and /events an endpoint of yours
with client.stream("GET", "/events") as response:
    first = next(response.iter_text())  # leaving the block stops the stream
```

To read a Server-Sent Events stream, `iter_events()` (`aiter_events()` on `AsyncTestClient`) yields `ServerSentEvent`s:

```
# /feed is an endpoint of yours that sends events
with client.stream("GET", "/feed") as response:
    event = next(response.iter_events())  # event.event, event.data, event.id
```

`client.websocket(path)` opens one: `send_text()`, `send_json()`..., `receive_text()`, `receive_json()`... A refused connection raises `WebSocketDisconnect`. It takes `query=`, `headers=`, `cookies=` and `subprotocols=["chat"]`; `ws.subprotocol` and `ws.headers` are those the app accepted with.

## Replacing a dependency

```
from korvex import App, Depends, Response
from korvex.testing import TestClient

app = App()


class User:  # your code
    def __init__(self, id: int, name: str):
        self.id, self.name = id, name


def current_user() -> User:  # your code: reads the session, a token...
    raise NotImplementedError


@app.get("/me")
def me(user: User = Depends(current_user)):
    return Response.json({"name": user.name})


app.dependency_overrides[current_user] = lambda: User(id=1, name="test")
```

## Async tests

`AsyncTestClient` does the same, awaited, for `async` tests (pytest-asyncio or anyio):

```
from korvex.testing import AsyncTestClient


async def test_live():  # app: the one above
    async with AsyncTestClient(app) as client:
        response = await client.get("/api/tasks")
        assert response.status == 401
        async with client.websocket("/live") as ws:
            assert (await ws.receive_json())["event"] == "presence"
```

Next: [The built-in server](https://3d3n-pyc.github.io/korvex/server/index.md).

# The built-in server

Use this when you run your app on Linux or macOS and want the fastest setup, without installing an [ASGI](https://3d3n-pyc.github.io/korvex/glossary/#asgi) server.

Korvex runs an app without an ASGI server, on Linux and macOS:

```
korvex run main:app --workers 3
```

or, from Python, `app.run(workers=3)`. On the same CPU cores, it serves 21 to 32% more requests than uvicorn, with 40% less memory (see [Performance](https://3d3n-pyc.github.io/korvex/performance/index.md)).

The app stays a plain [ASGI](https://3d3n-pyc.github.io/korvex/glossary/#asgi) app: uvicorn, Granian or a serverless platform run it too, and Windows needs one of them.

## Options

The same in `korvex run` and `app.run()`. In Python, use underscores and pass lists as lists:

```
app.run("0.0.0.0", 8000, workers=3, graceful_timeout=10, forwarded_allow_ips=["10.0.0.5"])
```

| Option                  | Default             | Does                                                                                                                                                                                                                                                                                                                                                                                                                                  |
| ----------------------- | ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `--host`, `--port`      | `127.0.0.1`, `8000` | Address to listen on.                                                                                                                                                                                                                                                                                                                                                                                                                 |
| `--uds PATH`            |                     | Listen on a Unix socket instead, for a proxy on the same machine.                                                                                                                                                                                                                                                                                                                                                                     |
| `--workers N`           | `1`                 | [Worker](https://3d3n-pyc.github.io/korvex/glossary/#worker) processes sharing the port; about 3 per 4 CPU cores works best. A worker that crashes is restarted.                                                                                                                                                                                                                                                                      |
| `--max-connections N`   | `0` (no limit)      | Open connections per worker and thread (WebSockets count). Beyond it, new clients wait in the listen backlog, not refused, until one closes. Memory is bounded by about N × `--max-body`.                                                                                                                                                                                                                                             |
| `--max-requests N`      | `0` (never)         | A worker stops gracefully after about N HTTP requests (N plus a random 0 to 10%, so workers don't restart together) and a new one replaces it at once: a cap on slow memory leaks. Counted per worker, all its threads together; WebSocket handshakes aren't counted.                                                                                                                                                                 |
| `--threads N`           | `1`                 | Event loops per worker, each in its own thread: see below.                                                                                                                                                                                                                                                                                                                                                                            |
| `--graceful-timeout S`  | `30`                | On Ctrl+C or SIGTERM, seconds requests in progress get to finish.                                                                                                                                                                                                                                                                                                                                                                     |
| `--max-body BYTES`      | 1 MiB               | Larger bodies get a 413, larger WebSocket messages close the connection. A route can set its own: `@app.post("/upload", max_body=50_000_000)`.                                                                                                                                                                                                                                                                                        |
| `--read-timeout S`      | `30`                | Seconds a client gets to send the headers, then the body. Slow headers (or an idle connection) are closed with no response; a slow body gets a 408.                                                                                                                                                                                                                                                                                   |
| `--forwarded-allow-ips` | `127.0.0.1,::1`     | Proxies whose `X-Forwarded-For` and `-Proto` (or `Forwarded`) give the real client and scheme. `*` trusts any.                                                                                                                                                                                                                                                                                                                        |
| `--forwarded-host`      | off                 | Also take the host they forward. Only if your proxy sets that header itself: nginx passes a client's `X-Forwarded-Host` on as it is.                                                                                                                                                                                                                                                                                                  |
| `--root-path PREFIX`    | none                | The [`root_path`](https://3d3n-pyc.github.io/korvex/glossary/#root_path): the prefix a proxy strips before forwarding (nginx `location /api/ { proxy_pass http://app/; }`), such as `/api`. Empty, or starting with `/` and not ending with one. The proxy strips the prefix, so the server routes the path it receives as is, never stripping again. It only sets `request.root_path` and puts the prefix back in links (`url_for`). |
| `--access-log`          | off                 | Log each request.                                                                                                                                                                                                                                                                                                                                                                                                                     |
| `--ws-ping-interval S`  | `20`                | Seconds between pings on open WebSockets; a client silent until the next ping is closed. `0` turns them off.                                                                                                                                                                                                                                                                                                                          |

## Threads

With `--threads 3`, each worker runs 3 event loops, in 3 threads. On free-threaded Python (3.14t) they run in parallel: one process uses several cores and shares its memory. With the GIL, they take turns on one core, so use workers instead.

Each [event loop](https://3d3n-pyc.github.io/korvex/glossary/#event-loop) runs the startup and shutdown hooks. Keep what they open in `app.local`, which holds one value per loop: a module global would give every loop the first loop's connection pool, and asyncio objects only work on the loop that made them.

## How it works

A Rust thread per loop reads each request in full, without touching Python, then hands it to the asyncio loop that runs your handlers. A slow or huge upload never holds Python up.

## Good to know

- `--max-requests` recycles workers: even with one worker, the server forks it and restarts it as a supervisor would. A clean restart never triggers the crash backoff.
- In production, run it behind a [reverse proxy](https://3d3n-pyc.github.io/korvex/glossary/#reverse-proxy) (nginx, Caddy, Traefik): it speaks HTTP/1.1 and WebSocket, and leaves HTTPS, HTTP/2 and HTTP/3 to the proxy. The [deployment guide](https://3d3n-pyc.github.io/korvex/deployment/index.md) has ready configurations.
- A body that's too big (413), too slow (408) or broken (400) is still answered by the app, so before hooks, exception handlers, after hooks and the access log see it.
- A slow client can't make a stream or a WebSocket pile up in memory: once 1 MiB waits for it, the handler's next send waits.
- A client can't flood a WebSocket handler: once 1 MiB of its messages waits for `receive()`, the server stops reading from it until the handler catches up. A send-only handler learns the client left from its next send.
- With several workers, a worker that crashes after starting is restarted after a growing delay (0.1 s, doubling up to 30 s). On SIGTERM or SIGINT the workers get `graceful_timeout` to finish, and are killed 5 s later if still there; a second signal kills them at once.
- From a proxy's forwarded headers, the rightmost value counts: the one the nearest proxy set, not one the client could have written.

Next: [Deploying a Korvex app](https://3d3n-pyc.github.io/korvex/deployment/index.md), with proxy configurations, Docker and Kubernetes.

# Deploying a Korvex app

On this page:

- [Production checklist](#production-checklist)
- [Default limits](#default-limits)
- [1. Built-in server behind a reverse proxy](#1-built-in-server-behind-a-reverse-proxy)
- [2. uvicorn behind a reverse proxy](#2-uvicorn-behind-a-reverse-proxy)
- [3. Docker](#3-docker)
- [4. AWS Lambda](#4-aws-lambda)
- [5. Vercel](#5-vercel)
- [Running in production](#running-in-production): keep-alive timeouts, Kubernetes, logging, health route, blocking code, file descriptors

An `App` runs on Korvex's built-in server (Linux and macOS) or on any ASGI server, including serverless platforms. The files are in [`examples/deploy/`](https://github.com/3d3n-pyc/korvex/tree/main/examples/deploy), around a small [`main.py`](https://github.com/3d3n-pyc/korvex/blob/main/examples/deploy/main.py).

| Setup                                                                               | When                                              | Checked                                                                            |
| ----------------------------------------------------------------------------------- | ------------------------------------------------- | ---------------------------------------------------------------------------------- |
| [Built-in server behind a reverse proxy](#1-built-in-server-behind-a-reverse-proxy) | Your own server or VM, Linux or macOS             | Yes, behind Caddy and nginx over a Unix socket, in containers, WebSockets included |
| [uvicorn behind a reverse proxy](#2-uvicorn-behind-a-reverse-proxy)                 | The same, or Windows                              | Yes, behind Caddy and nginx, in containers                                         |
| [Docker](#3-docker)                                                                 | Fly.io, Railway, Render, Cloud Run, Kubernetes... | Yes, image built and run, stops cleanly on SIGTERM                                 |
| [AWS Lambda](#4-aws-lambda)                                                         | Serverless on AWS                                 | Locally: handler called with an API Gateway event, not on AWS                      |
| [Vercel](#5-vercel)                                                                 | Serverless on Vercel                              | Yes, deployed and tested                                                           |

With your own server, put a reverse proxy in front: it handles HTTPS, HTTP/2 and HTTP/3, then forwards each request to Korvex in HTTP/1.1.

If the proxy serves the app under a path prefix (`example.com/api/`), give Korvex that prefix so links come out right: `--root-path /api` for uvicorn or the built-in server (`app.run(root_path="/api")`). See the [server guide](https://3d3n-pyc.github.io/korvex/server/index.md).

## Production checklist

Before you open the app to real traffic:

- Run it behind a [reverse proxy](https://3d3n-pyc.github.io/korvex/glossary/#reverse-proxy) that serves HTTPS, and tell Korvex which proxies to trust with `--forwarded-allow-ips`: [section 1](#1-built-in-server-behind-a-reverse-proxy).
- Set `--workers` to about 3 per 4 CPU cores (the default is 1): [section 1](#1-built-in-server-behind-a-reverse-proxy).
- Read secrets, such as the sessions `secret_key`, from the environment, never from the code: [Sessions](https://3d3n-pyc.github.io/korvex/sessions/index.md).
- Keep the sessions cookie `secure=True` (the default): it is sent over HTTPS only, so a plain `http://` setup needs the proxy's HTTPS first: [Sessions](https://3d3n-pyc.github.io/korvex/sessions/#what-a-session-is).
- Call `csrf(app)` if the app has HTML forms that use cookies: [CSRF](https://3d3n-pyc.github.io/korvex/sessions/#csrf-csrfapp).
- Call `security_headers(app)`: [Security headers](https://3d3n-pyc.github.io/korvex/middleware/#security-headers-security_headersapp).
- Configure logging, and turn on `--access-log` if you want one line per request: [Logging](#logging).
- Add a health route for the proxy or orchestrator: [A health route](#a-health-route).
- On Kubernetes, add a `preStop` sleep and a long enough grace period: [Kubernetes and containers](#kubernetes-and-containers).
- Keep blocking code out of plain `def` handlers, which run on the event loop: [Blocking code](#blocking-code).
- Make the proxy's idle timeout shorter than Korvex's: [Keep-alive timeouts](#keep-alive-timeouts).
- Review the [default limits](#default-limits) (body size, timeouts, connections) and raise or lower them for your app.
- Raise the file descriptor limit if you expect many clients: [File descriptors](#file-descriptors).

## Default limits

The built-in server's limits, with what a client sees when it goes over. Set them with the CLI flag, or the keyword of `app.run()`:

| Limit                                                             | Default                                        | The client sees                                                                                                                 |
| ----------------------------------------------------------------- | ---------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------- |
| Request body: `--max-body`, `max_body=` (a route can set its own) | 1 MiB (1,048,576 bytes)                        | `413`. A WebSocket message over it closes the connection.                                                                       |
| Read time: `--read-timeout`, `read_timeout=`                      | 30 seconds                                     | A body that takes longer gets a `408`. Headers that take longer, or an idle keep-alive connection, are closed with no response. |
| Request headers (not an option)                                   | More than 100 headers, or about 400 KB of them | `431`, and the connection closes.                                                                                               |
| Stop time: `--graceful-timeout`, `graceful_timeout=`              | 30 seconds                                     | After a SIGTERM or Ctrl+C, requests still running when it ends are cut off.                                                     |
| WebSocket ping: `--ws-ping-interval`, `ws_ping_interval=`         | 20 seconds (`0`: off)                          | A client that doesn't answer by the next ping is closed.                                                                        |
| Connections: `--max-connections`, `max_connections=`              | `0` (no limit), per worker and thread          | New clients wait in the listen backlog until a connection closes; they aren't refused.                                          |
| Worker life: `--max-requests`, `max_requests=`                    | `0` (never restart)                            | Nothing: the worker stops gracefully after about that many requests, and a new one replaces it.                                 |

The header limits come from the HTTP library the server uses, not from a Korvex option. A `413` or `408` is still answered by the app, so your exception handlers and the access log see it. All options are in the [server guide](https://3d3n-pyc.github.io/korvex/server/#options).

## 1. Built-in server behind a reverse proxy

The fastest setup (see the server comparison in the README). With the proxy on the same machine, use a Unix socket:

```
pip install --pre korvex
korvex run main:app --workers 3 --uds /run/korvex/korvex.sock
```

- `--workers`: about 3 per 4 CPU cores. Each worker is a separate process, restarted if it crashes.
- `--uds`: only processes on the machine can reach the app. The socket file is writable by every user, as with uvicorn, so a proxy running as another user can connect; restrict access with the permissions of its directory (`/run/korvex` here). Keep the path short: macOS allows 104 characters.
- Requests through the socket are trusted: `request.client` and `request.scheme` come from the proxy's `X-Forwarded-For` and `X-Forwarded-Proto`. `Host` is the one nginx sets (`proxy_set_header Host $host`) or Caddy passes on. Use `--forwarded-host` (read `X-Forwarded-Host`) only if your proxy sets that header; this nginx configuration doesn't, so it would pass a client's on.
- WebSockets get a ping every 20 seconds (`--ws-ping-interval`), which also keeps them open through nginx's 60-second `proxy_read_timeout`. A client that stops answering is closed.
- On SIGTERM (systemd, Docker, Kubernetes), requests in progress get up to 30 seconds to finish (`--graceful-timeout`).
- To listen on TCP instead, use `--host 127.0.0.1 --port 8000` with the proxy configurations of section 2.

With [Caddy](https://github.com/3d3n-pyc/korvex/blob/main/examples/deploy/Caddyfile-unix):

```
example.com {
    reverse_proxy unix//run/korvex/korvex.sock
}
```

With [nginx](https://github.com/3d3n-pyc/korvex/blob/main/examples/deploy/nginx-unix.conf):

```
# Passes WebSocket upgrades on, and keeps plain requests as they were.
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

upstream korvex {
    server unix:/run/korvex/korvex.sock;
}

server {
    listen 80;
    server_name example.com;

    location / {
        proxy_pass http://korvex;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}
```

Caddy passes WebSockets on by itself. nginx needs the `map` and the `Upgrade` / `Connection` headers above, or WebSocket connections reach the app as plain requests and fail.

Keep `korvex run` running with systemd, or use Docker (section 3).

## 2. uvicorn behind a reverse proxy

For Windows, where the built-in server isn't available, or to keep an ASGI server:

```
pip install --pre "korvex[serve]"
uvicorn main:app --host 127.0.0.1 --port 8000 --workers 4 \
    --proxy-headers --forwarded-allow-ips 127.0.0.1
```

- `--workers`: one per CPU core. Each worker is a separate process.
- `--proxy-headers` with `--forwarded-allow-ips`: uvicorn trusts the proxy's `X-Forwarded-*` headers, so `request.client`, `request.scheme` and its logs show the client's.
- Listen on `127.0.0.1`, not `0.0.0.0`, so only the proxy can reach the app.

With [Caddy](https://github.com/3d3n-pyc/korvex/blob/main/examples/deploy/Caddyfile), which gets and renews the HTTPS certificate itself:

```
example.com {
    reverse_proxy 127.0.0.1:8000
}
```

With [nginx](https://github.com/3d3n-pyc/korvex/blob/main/examples/deploy/nginx.conf), use the configuration of section 1 without the `upstream` block, and point `proxy_pass` at the TCP address:

```
        proxy_pass http://127.0.0.1:8000;
```

Add a certificate, for example with certbot, to serve HTTPS.

## 3. Docker

```
FROM python:3.14-slim
WORKDIR /app

# Korvex ships prebuilt wheels: no Rust toolchain needed.
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py .

# Run as an unprivileged user, not root.
RUN useradd --no-create-home --shell /usr/sbin/nologin app
USER app

EXPOSE 8000
CMD ["korvex", "run", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```

`requirements.txt` contains `korvex`. Wheels exist for Linux x86_64 and ARM64, so the image builds on both.

- Set `--workers` to about 3/4 of the container's CPUs (the default, 1, fits a single CPU).
- `docker stop` sends SIGTERM: requests in progress finish, then the container exits (in 0.1 s when idle, in our test).

Hosting platforms (Fly.io, Railway, Render, Cloud Run...) put their own proxy in front, handle HTTPS and add containers when traffic grows. On your own server, put Caddy or nginx in front as in section 1.

## 4. AWS Lambda

[Mangum](https://pypi.org/project/mangum/) turns Lambda events into ASGI calls. [`lambda_handler.py`](https://github.com/3d3n-pyc/korvex/blob/main/examples/deploy/lambda_handler.py):

```
from mangum import Mangum

from main import app

handler = Mangum(app)
```

Package `korvex` and `mangum` with your code, set the Lambda handler to `lambda_handler.handler`, and put an API Gateway (HTTP API) or a function URL in front.

**Startup hooks run on every request here.** Mangum replays the startup and shutdown events around each Lambda call, so `@app.on_startup` and `@app.on_shutdown` run every time. Don't open expensive resources there (a database pool): create them once at module level or on first use, so a warm Lambda instance reuses them.

## 5. Vercel

Vercel finds the app itself: put it in `main.py` (or `app.py`) as a top-level `app` variable and list `korvex` in `requirements.txt`, as `examples/deploy/main.py` does. Then deploy:

```
npx vercel deploy --prod
```

If Vercel doesn't find the app, point to it in `pyproject.toml`:

```
[tool.vercel]
entrypoint = "main:app"
```

Vercel runs the app as ASGI in its own runtime, so the server options above don't apply. Checked on a real deployment:

- Typed path parameters, cookies, forms (urlencoded and multipart, with a file), streaming, `HTTPException`, 404, 405 and `HEAD` behave as they do locally.
- **Startup hooks run once per instance**, not on every request (unlike Lambda with Mangum): 20 requests in a row saw one startup. Opening a database pool in `@app.on_startup` is fine here.
- Each new instance starts with an empty lookup cache.

## Running in production

Things that work in a quick test but break under real traffic.

### Keep-alive timeouts

Korvex closes a connection idle for `--read-timeout` (30 seconds by default). If the proxy keeps idle connections open longer, it can send a request on one Korvex just closed, and the client gets a sporadic 502. **The proxy must close idle connections before Korvex does.**

- **nginx.** The configurations above send `Connection: close` for plain requests, so nginx opens a new connection each time and this can't happen. If you turn on upstream keepalive, set `keepalive_timeout` below 30 seconds:

```
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      '';       # not "close": lets nginx reuse the connection
}

upstream korvex {
    server 127.0.0.1:8000;
    keepalive 32;
    keepalive_timeout 25s;
}
```

Keep `proxy_http_version 1.1;` and `Connection $connection_upgrade` in the `location` block.

- **Caddy.** It keeps idle upstream connections for 2 minutes. Lower it:

```
example.com {
  reverse_proxy 127.0.0.1:8000 {
      transport http {
          keepalive 25s
      }
  }
}
```

- **AWS Application Load Balancer.** Its idle timeout is 60 seconds by default. Set it below 30 seconds, or start Korvex with `--read-timeout 65` or more (which also gives slow clients longer to send their request).

### Kubernetes and containers

On SIGTERM, Korvex stops accepting connections at once, then gives requests in progress up to `--graceful-timeout` (30 seconds) to finish. Kubernetes removes the pod from its Service at the same moment, so the load balancer may still send it a few requests. A short `preStop` sleep keeps the pod serving until the balancer catches up. The grace period must cover the sleep and the graceful timeout:

```
spec:
  terminationGracePeriodSeconds: 45   # preStop 5 + graceful timeout 30 + margin
  containers:
    - name: app
      image: example/app:latest
      lifecycle:
        preStop:
          exec:
            command: ["sleep", "5"]
      readinessProbe:
        httpGet:
          path: /health
          port: 8000
```

The `sleep` command must exist in the image (`python:slim` has it).

### Logging

Korvex logs through Python's `logging`. Configure it once, at the top of `main.py`, and turn on the access log (one line per request, to the `korvex.access` logger) when you start the server:

```
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
```

```
korvex run main:app --workers 3 --access-log
```

Without `--access-log`, requests aren't logged. If logging is already configured when it starts, Korvex doesn't add its own handler for the access log, so your format applies. In a container, log to stdout and let the platform collect it.

### Blocking code

A plain `def` handler runs on the [event loop](https://3d3n-pyc.github.io/korvex/glossary/#event-loop), not in a thread. A blocking call in it (a synchronous database driver, `requests`, `time.sleep`) holds up every other request of that [worker](https://3d3n-pyc.github.io/korvex/glossary/#worker) until it returns. Use async libraries, or move the blocking part to a thread:

```
import asyncio

@app.get("/report")
async def report():
    data = await asyncio.to_thread(build_report)  # build_report blocks
    return Response.json(data)
```

More workers help, but they don't fix a handler that blocks for seconds.

### A health route

Proxies and orchestrators need a URL that answers 200 when the app is up:

```
@app.get("/health", include_in_schema=False)
def health():
    return Response.json({"ok": True})
```

### File descriptors

Each open connection, WebSockets included, uses one file descriptor, and the default limit is often 1024 per process. Past it, new connections fail. Raise it if you expect many clients:

- systemd: `LimitNOFILE=65536` in the `[Service]` section.
- Docker: `docker run --ulimit nofile=65536:65536 ...`.
# Reference

# The App

`App` is the application: routes, hooks, errors, OpenAPI, and `run()` for the built-in server. `app.group()` and `app.host()` return a `Group`. The [guide](https://3d3n-pyc.github.io/korvex/routes/index.md) shows how they fit together; this page lists every method and option.

## App

```
App()
```

The application: routes, hooks, error handlers, OpenAPI, and `run()`. It is an ASGI app.

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

### local

```
local = LoopLocal()
```

Attributes per event loop: see `LoopLocal`.

### dependency_overrides

```
dependency_overrides: dict[Callable[..., Any], Callable[..., Any]] = {}
```

`{dependency: replacement}`: called instead of the dependency, for tests.

### get

```
get(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `GET path`, with the options of `App.route()`.

### post

```
post(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `POST path`, with the options of `App.route()`.

### put

```
put(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `PUT path`, with the options of `App.route()`.

### patch

```
patch(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `PATCH path`, with the options of `App.route()`.

### delete

```
delete(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `DELETE path`, with the options of `App.route()`.

### head

```
head(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `HEAD path`, with the options of `App.route()`.

### options

```
options(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `OPTIONS path`, with the options of `App.route()`.

### route

```
route(path: str, *, method: str = 'GET', name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `path` for `method`. `@app.get(path)` and the others take the same options.

- `name`: what `url_for()` knows the route by (the function's name by default).
- `status`: the status of a response made from a model or `None` (`-> None` with `204`: empty).
- `dependencies`: `[Depends(func), ...]`, run before the handler.
- `tags`, `deprecated`, `responses`, `include_in_schema`: the route's OpenAPI entry.
- `max_body` (bytes): the built-in server's body limit for this route; ASGI servers ignore it.

Guide: https://3d3n-pyc.github.io/korvex/routes/#route-options

### group

```
group(prefix: str = '', *, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True) -> Group
```

Returns a `Group` that registers handlers under `prefix`, with `dependencies` and OpenAPI options for all.

Guide: https://3d3n-pyc.github.io/korvex/routes/#groups

### host

```
host(pattern: str, *, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True) -> Group
```

Returns a group whose routes only answer for requests to `pattern`, a host (`{param}` takes one label).

Calling it again with the same pattern adds to the same routes; a route's `max_body` applies to its path on every host. Guide: https://3d3n-pyc.github.io/korvex/routes/#hosts

### websocket

```
websocket(path: str, *, name: str | None = None, dependencies: Sequence[Any] = ()) -> Callable[[WebSocketHandler], WebSocketHandler]
```

Decorator: registers the function as the WebSocket handler for `path`.

It gets a `WebSocket`, path params, and parameters and dependencies as HTTP handlers do (no body). `name`: for `url_for()`. Guide: https://3d3n-pyc.github.io/korvex/websockets/

### url_for

```
url_for(name: str, /, **params: Any) -> str
```

The path of the route named `name`, with `params` for its path parameters.

Raises LookupError if several routes share a function's name (give a `name=`; routes share a `name=` only on one path, for different methods). Raises ValueError for a value that doesn't fit its constraint or the handler's annotation, is empty, `.` or `..`, or has a `/` outside a `*wildcard`, or when another route answers the path first (TypeError for an unsupported type). Every path parameter must be given, and nothing else: add a query string yourself. Values are percent-encoded; a host route gives the path only. An app is served under one `root_path`: it prefixes the path. Guide: https://3d3n-pyc.github.io/korvex/routes/#links-to-a-route

### openapi

```
openapi(*, title: str = 'API', version: str = '0.1.0', description: str | None = None, host: str | None = None) -> dict
```

An OpenAPI 3.1 description of the app's routes (static files and WebSockets left out).

With `host`, the routes that answer there; without, only those with no host. Guide: https://3d3n-pyc.github.io/korvex/openapi/

### docs

```
docs(*, title: str = 'API', version: str = '0.1.0', description: str | None = None, openapi_path: str = '/openapi.json', docs_path: str | None = '/docs') -> None
```

Serves the OpenAPI document at `openapi_path` and Swagger UI at `docs_path` (`None`: none).

`title`, `version` and `description` go in the document. Guide: https://3d3n-pyc.github.io/korvex/openapi/

### constraint

```
constraint(name: str, pattern: str) -> None
```

Defines a route constraint `{param:name}`, matching the whole value against `pattern`.

Rust `regex` syntax. Constrained parameters are tried in registration order, before plain ones; the handler gets a `str`. Define it before the routes that use it; it can't be redefined differently. Guide: https://3d3n-pyc.github.io/korvex/routes/#your-own-constraints

### static

```
static(prefix: str, directory: str | PathLike[str], *, name: str | None = None) -> None
```

Serves the files in `directory` under `prefix`; paths leaving it get a 404.

`name` (`"static"` by default) is what `url_for()` knows it by. Guide: https://3d3n-pyc.github.io/korvex/routes/#static-files

### mount

```
mount(prefix: str, asgi_app: Callable, *, name: str | None = None) -> None
```

Serves another ASGI app (a Starlette app, a Korvex `App`...) under `prefix`.

The app sees paths without `prefix` (its `root_path` has it). A route of this app that matches wins. This app's before hooks run first (they protect the mount); its after hooks and exception handlers don't run for the mount. The mounted app's lifespan runs with this app's startup and shutdown. The bare `prefix` redirects (308) to `prefix/`. `name`: `url_for(name)` gives `prefix/`, `url_for(name, path="a/b")` gives `prefix/a/b`. Raises ValueError for an invalid or already mounted `prefix`. Guide: https://3d3n-pyc.github.io/korvex/routes/#mounting-another-app

### static_url

```
static_url(path: str, /, *, name: str | None = None) -> str
```

The URL of a static file, with a hash of its content: `/static/app.3f2a1c9b.css`.

`path`: the file, relative to the folder given to `static()`. `name`: which `static()` mount, when there are several (`"static"` by default). Raises LookupError for an unknown mount, FileNotFoundError for a missing file. An app is served under one `root_path`: it prefixes the URL. Guide: https://3d3n-pyc.github.io/korvex/templates/

### templates

```
templates(directory: str | PathLike[str] = 'templates', *, reload: bool = False) -> Templates
```

Returns the `Templates` for `directory`, with `url_for`, `static_url` and `get_flashed_messages` available in every template.

`directory`: the template folder, relative to the current directory. `reload`: recompile templates when their files change. Guide: https://3d3n-pyc.github.io/korvex/templates/

### on_startup

```
on_startup(func: Hook) -> Hook
```

```
on_startup(func: None = None) -> Callable[[Hook], Hook]
```

```
on_startup(func: Hook | None = None) -> Hook | Callable[[Hook], Hook]
```

Decorator: runs the function at server start; if it raises, the server doesn't start.

Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/#startup-and-shutdown

### on_shutdown

```
on_shutdown(func: Hook) -> Hook
```

```
on_shutdown(func: None = None) -> Callable[[Hook], Hook]
```

```
on_shutdown(func: Hook | None = None) -> Hook | Callable[[Hook], Hook]
```

Decorator: runs the function at server stop, in reverse order of declaration.

Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/#startup-and-shutdown

### before_request

```
before_request(func: BeforeHook) -> BeforeHook
```

```
before_request(func: None = None) -> Callable[[BeforeHook], BeforeHook]
```

```
before_request(func: BeforeHook | None = None) -> BeforeHook | Callable[[BeforeHook], BeforeHook]
```

Decorator: runs the function before routing, so also for unknown paths and handshakes.

It gets the request (`path_params` empty) and returns a `Response` to answer, or `None`. For a WebSocket handshake, answering or raising `HTTPException` or `Abort` refuses it. Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/#hooks-around-every-request

### after_request

```
after_request(func: AfterHook) -> AfterHook
```

```
after_request(func: None = None) -> Callable[[AfterHook], AfterHook]
```

```
after_request(func: AfterHook | None = None) -> AfterHook | Callable[[AfterHook], AfterHook]
```

Decorator: runs the function on every response, errors included, in reverse order.

It returns the response to send, or `None` to keep it. Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/#hooks-around-every-request

### exception_handler

```
exception_handler(key: int | type[Exception]) -> Callable[[ExceptionHandler], ExceptionHandler]
```

Decorator: answers errors with the function's `Response`.

- `key`: an HTTP status (`404` also covers unknown paths) or an exception class.

Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/#answering-errors-your-way

### run

```
run(host: str = '127.0.0.1', port: int = 8000, *, workers: int = 1, graceful_timeout: float = 30.0, max_body: int = 1024 * 1024, read_timeout: float = 30.0, uds: str | None = None, forwarded_allow_ips: Sequence[str] = ('127.0.0.1', '::1'), access_log: bool = False, ws_ping_interval: float = 20.0, forwarded_host: bool = False, threads: int = 1, max_connections: int = 0, max_requests: int = 0, root_path: str = '') -> None
```

Serves the app with the built-in server until Ctrl+C or SIGTERM (Linux and macOS only).

- `workers`: processes sharing the port, each running the startup and shutdown hooks.
- `threads`: event loops per worker; keep what they open in `app.local`.
- `graceful_timeout`: seconds requests in progress get on stop.
- `max_body`: bytes; larger bodies get a 413 (a route's `max_body` overrides it).
- `read_timeout`: seconds a client gets to send headers, then body (408).
- `uds`: a Unix socket path to listen on instead of host:port.
- `forwarded_allow_ips`: proxies trusted for `X-Forwarded-For`/`-Proto` (`"*"`: any).
- `forwarded_host`: also take the forwarded host, only if your proxy sets it itself.
- `access_log`: log each request to `korvex.access`.
- `ws_ping_interval`: seconds between WebSocket pings; `0` turns them off.
- `max_connections`: open connections per worker and thread; `0`: no limit.
- `max_requests`: a worker restarts after about that many requests; `0`: never.
- `root_path`: the prefix a proxy serves the app under (`"/api"`), for links.

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

## Group

```
Group(app: App, routes: RouteGroup, dependencies: Sequence[Any] = (), host: _Host | None = None, docs: _Docs | None = None)
```

Registers routes under a shared prefix, dependencies and OpenAPI options.

Guide: https://3d3n-pyc.github.io/korvex/routes/#groups

### get

```
get(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `GET path`, with the options of `App.route()`.

### post

```
post(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `POST path`, with the options of `App.route()`.

### put

```
put(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `PUT path`, with the options of `App.route()`.

### patch

```
patch(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `PATCH path`, with the options of `App.route()`.

### delete

```
delete(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `DELETE path`, with the options of `App.route()`.

### head

```
head(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `HEAD path`, with the options of `App.route()`.

### options

```
options(path: str, *, name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: the function answers `OPTIONS path`, with the options of `App.route()`.

### route

```
route(path: str, *, method: str = 'GET', name: str | None = None, status: int = 200, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True, max_body: int | None = None) -> Callable[[Handler], Handler]
```

Decorator: registers the handler for `path` under the group's prefix.

Guide: https://3d3n-pyc.github.io/korvex/routes/#route-options

### websocket

```
websocket(path: str, *, name: str | None = None, dependencies: Sequence[Any] = ()) -> Callable[[WebSocketHandler], WebSocketHandler]
```

Decorator: registers the WebSocket handler for `path` under the group's prefix.

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

### group

```
group(prefix: str = '', *, dependencies: Sequence[Any] = (), tags: Sequence[str] = (), deprecated: bool = False, responses: Mapping[int | str, Any] | None = None, include_in_schema: bool = True) -> Group
```

Returns a nested group, adding its prefix, dependencies and OpenAPI options.

## LoopLocal

```
LoopLocal()
```

Attributes kept per event loop: `app.local`, for what a startup hook opens (a pool, a client).

Use it from a running loop only. Guide: https://3d3n-pyc.github.io/korvex/server/#threads

# Errors

Raise them to answer with an error, or with an exact response. See [Errors and hooks](https://3d3n-pyc.github.io/korvex/errors-and-hooks/index.md).

## HTTPException

```
HTTPException(status: int, detail: str | None = None, *, headers: HeaderItems = ())
```

Raise it to answer with an HTTP error: `{"detail": ...}` as JSON.

- `detail`: the message; the standard reason if omitted
- `headers`: extra response headers, a dict or pairs

Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/

## RequestValidationError

```
RequestValidationError(errors: list[dict])
```

Raised when a request doesn't match a handler's annotations: a 422 with `errors`.

Guide: https://3d3n-pyc.github.io/korvex/validation/#errors

## Abort

```
Abort(response: Response)
```

Raise it to stop the request with this exact response, past exception handlers.

After hooks still apply. In a WebSocket's dependencies, it refuses the connection (403). Guide: https://3d3n-pyc.github.io/korvex/errors-and-hooks/

# korvex.middleware

Ready-made hooks. See [Ready-made middleware](https://3d3n-pyc.github.io/korvex/middleware/index.md) and [Sessions and CSRF](https://3d3n-pyc.github.io/korvex/sessions/index.md).

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

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

# Parameters

Where a parameter comes from when its annotation alone doesn't say. See [Validation](https://3d3n-pyc.github.io/korvex/validation/index.md) and [Dependencies](https://3d3n-pyc.github.io/korvex/dependencies/index.md).

## Query

```
Query(default: Any = REQUIRED, *, alias: str | None = None) -> Any
```

From the query string; `alias` gives a name that isn't a Python identifier.

Guide: https://3d3n-pyc.github.io/korvex/validation/#headers-cookies-and-forms

## Header

```
Header(default: Any = REQUIRED, *, alias: str | None = None) -> Any
```

From a request header (`x_client` reads `X-Client`); `alias` gives the exact name.

A `list[str]` gets every value of a repeated header. Guide: https://3d3n-pyc.github.io/korvex/validation/#headers-cookies-and-forms

## Cookie

```
Cookie(default: Any = REQUIRED, *, alias: str | None = None) -> Any
```

From a cookie of that name (or `alias`).

Guide: https://3d3n-pyc.github.io/korvex/validation/#headers-cookies-and-forms

## Form

```
Form(default: Any = REQUIRED, *, alias: str | None = None) -> Any
```

From a form field, urlencoded or multipart; annotated `UploadedFile`, a file.

Annotated with a msgspec or pydantic model, each of its fields is read from the form field of that name. Guide: https://3d3n-pyc.github.io/korvex/validation/#headers-cookies-and-forms

## Depends

```
Depends(dependency: Callable[..., Any], *, cache: bool = True) -> Any
```

The value `dependency` returns (or yields), computed before the handler runs.

- `cache`: call it once per request; `False` calls it for each parameter

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

# Requests

What a handler can ask for: the `Request` (an HTTP request), the `WebSocket`, and what both share, `Connection`. See [Requests and responses](https://3d3n-pyc.github.io/korvex/requests-and-responses/index.md).

## Connection

```
Connection(scope: dict, receive: Callable[[], Awaitable[dict]], path_params: Mapping[str, Any], app: App | None = None)
```

What an HTTP request and a WebSocket share: path, headers, cookies, query, client, scheme.

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/

### method

```
method: str = scope.get('method', 'GET')
```

The HTTP method; `GET` for a WebSocket's handshake.

### path

```
path: str = scope['path']
```

The path the app routes on, without `root_path`.

### client

```
client: tuple[str, int] | None = tuple(client) if client else None
```

`(host, port)` of the client, behind a trusted proxy too; `None` if unknown.

### scheme

```
scheme: str = scope.get('scheme', 'http')
```

`"https"` or `"http"` (`"wss"` or `"ws"` for a WebSocket), as the client used it.

### app

```
app = app
```

The `App` handling it.

### root_path

```
root_path: str
```

The prefix the app is served under (ASGI `root_path`); `""` if none.

### query

```
query: MultiDict[str]
```

The query string's fields, parsed on first use.

### cookies

```
cookies: dict[str, str]
```

The cookies sent by the client, parsed on first use.

### url_for

```
url_for(name: str, /, **params: Any) -> str
```

The full URL of the route named `name`, with the request's scheme and host.

The `Host` header comes from the client: for an emailed link, use a configured base URL. Guide: https://3d3n-pyc.github.io/korvex/routes/#links-to-a-route

## Request

```
Request(scope: dict, receive: Callable[[], Awaitable[dict]], path_params: Mapping[str, Any], app: App | None = None)
```

An incoming HTTP request: a `Connection`, and its body.

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/

### method

```
method: str = scope.get('method', 'GET')
```

The HTTP method; `GET` for a WebSocket's handshake.

### path

```
path: str = scope['path']
```

The path the app routes on, without `root_path`.

### client

```
client: tuple[str, int] | None = tuple(client) if client else None
```

`(host, port)` of the client, behind a trusted proxy too; `None` if unknown.

### scheme

```
scheme: str = scope.get('scheme', 'http')
```

`"https"` or `"http"` (`"wss"` or `"ws"` for a WebSocket), as the client used it.

### app

```
app = app
```

The `App` handling it.

### root_path

```
root_path: str
```

The prefix the app is served under (ASGI `root_path`); `""` if none.

### query

```
query: MultiDict[str]
```

The query string's fields, parsed on first use.

### cookies

```
cookies: dict[str, str]
```

The cookies sent by the client, parsed on first use.

### url_for

```
url_for(name: str, /, **params: Any) -> str
```

The full URL of the route named `name`, with the request's scheme and host.

The `Host` header comes from the client: for an emailed link, use a configured base URL. Guide: https://3d3n-pyc.github.io/korvex/routes/#links-to-a-route

### body

```
body() -> bytes
```

Reads the whole body, once; later calls return it again.

### json

```
json() -> JSONValue
```

Reads the body as JSON; invalid JSON answers a 400.

### form

```
form() -> FormData
```

Reads the body as a urlencoded or multipart form; files are `UploadedFile`s.

Any other content type answers a 415, undecodable text a 400.

## BackgroundTasks

```
BackgroundTasks()
```

Work to do after the response is sent, in the order it was added.

Async functions are awaited; plain functions run in a thread. Errors are logged, and the client never sees them.

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/#background-tasks

### add

```
add(func: Callable[..., Any], /, *args: Any, **kwargs: Any) -> None
```

Queues `func(*args, **kwargs)` to run after the response.

## WebSocket

```
WebSocket(scope: dict, receive: Callable, send: Callable, path_params: Mapping[str, Any], app: App | None = None)
```

A WebSocket connection, passed to `@app.websocket` handlers.

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

### method

```
method: str = scope.get('method', 'GET')
```

The HTTP method; `GET` for a WebSocket's handshake.

### path

```
path: str = scope['path']
```

The path the app routes on, without `root_path`.

### client

```
client: tuple[str, int] | None = tuple(client) if client else None
```

`(host, port)` of the client, behind a trusted proxy too; `None` if unknown.

### scheme

```
scheme: str = scope.get('scheme', 'http')
```

`"https"` or `"http"` (`"wss"` or `"ws"` for a WebSocket), as the client used it.

### app

```
app = app
```

The `App` handling it.

### root_path

```
root_path: str
```

The prefix the app is served under (ASGI `root_path`); `""` if none.

### query

```
query: MultiDict[str]
```

The query string's fields, parsed on first use.

### cookies

```
cookies: dict[str, str]
```

The cookies sent by the client, parsed on first use.

### subprotocols

```
subprotocols: list[str] = list(scope.get('subprotocols', []))
```

Subprotocols the client asked for, to pick one in `accept()`.

### url_for

```
url_for(name: str, /, **params: Any) -> str
```

The full URL of the route named `name`, with the request's scheme and host.

The `Host` header comes from the client: for an emailed link, use a configured base URL. Guide: https://3d3n-pyc.github.io/korvex/routes/#links-to-a-route

### accept

```
accept(subprotocol: str | None = None, *, headers: HeaderItems = ()) -> None
```

Accepts the connection, optionally with one of `subprotocols`.

### receive

```
receive() -> str | bytes
```

The next message, text or binary; raises `WebSocketDisconnect` once the client has gone.

### close

```
close(code: int = 1000, reason: str = '') -> None
```

Closes the connection. Before `accept()`, it refuses it (403).

## WebSocketDisconnect

```
WebSocketDisconnect(code: int = 1000)
```

Raised by `WebSocket.receive()` once the client has disconnected.

## UploadedFile

```
UploadedFile(filename: str, content_type: str, data: bytes)
```

A file sent in a `multipart/form-data` form.

`request.query` and `await request.form()` are a `MultiDict`:

## MultiDict

```
MultiDict(lists: dict[str, list[_V]] | None = None)
```

Names with one or more values, in order: `[name]` and `get()` give the first, `get_all()` all.

### get_all

```
get_all(name: str) -> list[_V]
```

Every value of `name`, in order; `[]` if there's none.

### multi_items

```
multi_items() -> list[tuple[str, _V]]
```

Every `(name, value)`, a name repeated for each of its values.

# Responses

What a handler returns: a `Response`, made directly or with its shortcuts (`Response.json()`, `Response.text()`...). See [Requests and responses](https://3d3n-pyc.github.io/korvex/requests-and-responses/#responses).

## Response

```
Response(body: str | bytes = b'', *, status: int = 200, headers: HeaderItems = ())
```

An HTTP response; a `str` body is encoded to UTF-8, so `body` is `bytes`.

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/#responses

### text

```
text(text: str, *, status: int = 200, headers: HeaderItems = ()) -> Response
```

A `text/plain` response (unless `headers` sets Content-Type).

### json

```
json(data: Any, *, status: int = 200, headers: HeaderItems = ()) -> Response
```

An `application/json` response (unless `headers` sets Content-Type); `data` is anything `json.dumps` takes.

### redirect

```
redirect(location: str, *, status: int = 302, headers: HeaderItems = ()) -> Response
```

An empty redirect to `location`.

### set_cookie

```
set_cookie(name: str, value: str, *, max_age: Optional[int] = None, path: str = '/', domain: Optional[str] = None, secure: bool = False, httponly: bool = False, samesite: Optional[Literal['lax', 'strict', 'none']] = 'lax') -> None
```

Adds a `Set-Cookie` header; `max_age` is in seconds (none: until the browser closes).

### delete_cookie

```
delete_cookie(name: str, *, path: str = '/', domain: Optional[str] = None) -> None
```

Tells the browser to remove a cookie (same `path` and `domain` as when set).

### stream

```
stream(chunks: Chunks, *, status: int = 200, headers: HeaderItems = ()) -> StreamingResponse
```

A response sent piece by piece, from a sync or async iterable of `bytes` or `str`.

### sse

```
sse(events: Union[Iterable[Any], AsyncIterable[Any]], *, ping: float = 15.0, status: int = 200, headers: HeaderItems = ()) -> StreamingResponse
```

A Server-Sent Events stream, for the browser's `EventSource`.

Args: events: A sync or async iterable of `ServerSentEvent`, or of plain values (sent as the data of an event). ping: Seconds of silence before a keep-alive comment is sent, so proxies keep the connection open. 0 turns it off. Only async iterables get pings: a sync generator is streamed as is. status: The HTTP status. headers: Extra headers; they win over the defaults.

```
async def clock():
    while True:
        yield ServerSentEvent({"time": time.time()}, event="tick")
        await asyncio.sleep(1)
return Response.sse(clock())
```

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/#server-sent-events

### file

```
file(path: str | PathLike[str], *, filename: Optional[str] = None, content_type: Optional[str] = None, status: int = 200, headers: HeaderItems = (), request: Optional[Request] = None) -> StreamingResponse
```

Sends a file in chunks, type guessed from its name; `filename` makes it a download.

`request`: pass it to answer conditional and range requests: 304, 206. Guide: https://3d3n-pyc.github.io/korvex/routes/#static-files

## Headers

```
Headers(items: HeaderItems = ())
```

HTTP headers: ordered, case-insensitive, repeated names allowed.

### get

```
get(name: str, default: str | None = None) -> str | None
```

The first value for `name`, or `default`.

### get_all

```
get_all(name: str) -> list[str]
```

Every value for `name`, in order.

### set

```
set(name: str, value: str) -> None
```

Replaces every value for `name` with `value`.

### remove

```
remove(name: str) -> None
```

Removes every value for `name`.

### add

```
add(name: str, value: str) -> None
```

Adds a value for `name`, keeping existing ones.

## ServerSentEvent

```
ServerSentEvent(data: Any = None, *, event: Optional[str] = None, id: Optional[str] = None, retry: Optional[int] = None)
```

One event of a Server-Sent Events stream.

Send it from a streaming handler; the browser's `EventSource` receives it.

Args: data: The payload. A str is sent as is; anything else is sent as JSON. None sends no data. event: The event name, a single line. Listeners use it to pick events. id: The event id, a single line without NUL. The browser sends the last one back on reconnect. retry: How many milliseconds the browser waits before reconnecting, an int of 0 or more.

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/#server-sent-events

Validate the fields, then build the wire bytes once so errors surface here.

### encode

```
encode() -> bytes
```

The event as UTF-8 bytes, ending with a blank line.

## StreamingResponse

```
StreamingResponse(chunks: Chunks, *, status: int = 200, headers: HeaderItems = ())
```

A response whose body is sent piece by piece: see `Response.stream()` and `Response.file()`.

### text

```
text(text: str, *, status: int = 200, headers: HeaderItems = ()) -> Response
```

A `text/plain` response (unless `headers` sets Content-Type).

### json

```
json(data: Any, *, status: int = 200, headers: HeaderItems = ()) -> Response
```

An `application/json` response (unless `headers` sets Content-Type); `data` is anything `json.dumps` takes.

### redirect

```
redirect(location: str, *, status: int = 302, headers: HeaderItems = ()) -> Response
```

An empty redirect to `location`.

### set_cookie

```
set_cookie(name: str, value: str, *, max_age: Optional[int] = None, path: str = '/', domain: Optional[str] = None, secure: bool = False, httponly: bool = False, samesite: Optional[Literal['lax', 'strict', 'none']] = 'lax') -> None
```

Adds a `Set-Cookie` header; `max_age` is in seconds (none: until the browser closes).

### delete_cookie

```
delete_cookie(name: str, *, path: str = '/', domain: Optional[str] = None) -> None
```

Tells the browser to remove a cookie (same `path` and `domain` as when set).

### stream

```
stream(chunks: Chunks, *, status: int = 200, headers: HeaderItems = ()) -> StreamingResponse
```

A response sent piece by piece, from a sync or async iterable of `bytes` or `str`.

### sse

```
sse(events: Union[Iterable[Any], AsyncIterable[Any]], *, ping: float = 15.0, status: int = 200, headers: HeaderItems = ()) -> StreamingResponse
```

A Server-Sent Events stream, for the browser's `EventSource`.

Args: events: A sync or async iterable of `ServerSentEvent`, or of plain values (sent as the data of an event). ping: Seconds of silence before a keep-alive comment is sent, so proxies keep the connection open. 0 turns it off. Only async iterables get pings: a sync generator is streamed as is. status: The HTTP status. headers: Extra headers; they win over the defaults.

```
async def clock():
    while True:
        yield ServerSentEvent({"time": time.time()}, event="tick")
        await asyncio.sleep(1)
return Response.sse(clock())
```

Guide: https://3d3n-pyc.github.io/korvex/requests-and-responses/#server-sent-events

### file

```
file(path: str | PathLike[str], *, filename: Optional[str] = None, content_type: Optional[str] = None, status: int = 200, headers: HeaderItems = (), request: Optional[Request] = None) -> StreamingResponse
```

Sends a file in chunks, type guessed from its name; `filename` makes it a download.

`request`: pass it to answer conditional and range requests: 304, 206. Guide: https://3d3n-pyc.github.io/korvex/routes/#static-files

# korvex.router

The standalone router, without the App. See [The standalone router](https://3d3n-pyc.github.io/korvex/router/index.md).

The standalone router: maps a path and method to a handler name.

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

## Router

```
Router()
```

Maps a path and method to a handler name; never calls your code.

`match` caches the last `CACHE_SIZE` lookups; results are read-only.

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

### version

```
version = 0
```

Goes up each time a route is added or removed.

### group

```
group(prefix: str = '', *, middleware: Sequence[str] | None = None) -> RouteGroup
```

Returns a `RouteGroup` that registers routes under `prefix`.

## RouteGroup

```
RouteGroup(router: Router, prefix: str = '', *, middleware: Sequence[str] | None = None)
```

Registers routes on a router under a prefix and middleware, from `Router.group()`.

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

### route

```
route(path: str, handler: str, *, method: str = 'GET', middleware: Sequence[str] | None = None) -> None
```

Registers a route. Its own `middleware` runs after the group's.

### full_path

```
full_path(path: str) -> str
```

`path` with this group's prefix: `/api` + `users` -> `/api/users`.

### group

```
group(prefix: str = '', *, middleware: Sequence[str] | None = None) -> RouteGroup
```

Returns a nested group. Its `middleware` runs after this group's.

## MethodNotAllowed

The path exists, but not for this method (HTTP 405).

### allowed

```
allowed: list[str]
```

The methods registered for this path.

# korvex.security

Sign-in schemes that guard routes and show in OpenAPI. See [OpenAPI and security](https://3d3n-pyc.github.io/korvex/openapi/#signing-in-from-swagger-ui).

Security schemes: dependencies that read a credential and show in OpenAPI.

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

## Bearer

```
Bearer(*, format: str | None = None, name: str = 'bearer', description: str | None = None, required: bool = True)
```

A token in `Authorization: Bearer <token>`; `format` documents it (`"JWT"`).

## APIKey

```
APIKey(*, header: str | None = None, query: str | None = None, cookie: str | None = None, name: str = 'api_key', description: str | None = None, required: bool = True)
```

A key in a header, query parameter or cookie: give exactly one (`TypeError` otherwise).

A query key ends up in access logs and browser history: prefer a header.

## OAuth2Password

```
OAuth2Password(token_url: str, *, scopes: Mapping[str, str] | None = None, refresh_url: str | None = None, name: str = 'oauth2', description: str | None = None, required: bool = True)
```

OAuth2 password flow: Swagger UI posts the form to `token_url`, then sends the bearer token.

`scopes` maps scope to description; `refresh_url` is optional.

## SecurityScheme

```
SecurityScheme(*, name: str, description: str | None, required: bool)
```

What the schemes share: `Depends(scheme)` gives the credential, a `str`.

- `name`: its key in OpenAPI's `securitySchemes` (a duplicate gets `name_2`...)
- `description`: shown in Swagger UI
- `required`: without a credential, answer 401 with `WWW-Authenticate` (False gives None)

Guide: https://3d3n-pyc.github.io/korvex/openapi/#signing-in-from-swagger-ui

### credential

```
credential(request: Connection) -> str | None
```

The credential the request carries, if any.

### openapi

```
openapi() -> dict[str, Any]
```

Its Security Scheme Object.

# Templates

`app.templates()` returns a `Templates`. The [guide](https://3d3n-pyc.github.io/korvex/templates/index.md) shows the syntax and the filters; this page lists the classes.

## Templates

```
Templates(directory: Union[str, Path], *, reload: bool = False)
```

Render HTML templates from a folder.

Templates are compiled the first time they are used, then cached.

```
templates = Templates("templates")
html = templates.render("index.html", {"user": "Ada"})
```

Set `reload=True` to pick up edited files. The guide: https://3d3n-pyc.github.io/korvex/templates/

Serve templates from `directory`; `reload` recompiles a template when its files change.

### render

```
render(name: str, context: Optional[Mapping[str, Any]] = None) -> str
```

Render a template and return the text.

`name` is the template's path in the folder. `context` is a dict (or any mapping) of the template's variables.

### response

```
response(name: str, context: Optional[Mapping[str, Any]] = None, /, *, status: int = 200, headers: HeaderItems = ()) -> Response
```

Render a template now and return it as an HTML response.

`name` is the template's path in the folder. `context` is a dict (or any mapping) of the template's variables. `status` is the HTTP status code. `headers` are extra headers; a Content-Type in them wins.

```
views.response("post.html", {"post": post})
```

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

## Markup

Text known to be safe HTML, so escaping leaves it as is; any str operation on it gives a plain str.

## TemplateError

Base class of every template error.

## TemplateSyntaxError

```
TemplateSyntaxError(message: str, name: str, line: int)
```

The template is invalid; `str()` reads `name, line N: message`.

## UndefinedError

A template used a name, key or attribute that does not exist.

# korvex.testing

The test client. See [Testing](https://3d3n-pyc.github.io/korvex/testing/index.md).

An in-process test client for any ASGI app: no server, no network.

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

## TestClient

```
TestClient(app: Callable[..., Awaitable[None]], *, base_url: str = 'http://testserver', headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, raise_server_errors: bool = True, timeout: float | None = 5.0)
```

Calls an ASGI app in-process, from regular tests.

Same arguments as `AsyncTestClient`. As a context manager it runs the app's lifespan. Raises `RuntimeError` inside a running event loop: use `AsyncTestClient` there.

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

### cookies

```
cookies: dict[str, str]
```

The cookies sent with each request: those the app set and any added here.

### headers

```
headers: Headers
```

Headers sent with each request.

### close

```
close() -> None
```

Closes the event loop, cancelling what the app left running.

### request

```
request(method: str, url: str, **kwargs: Any) -> TestResponse
```

Sends a request and returns the whole response; see `AsyncTestClient.request()`.

### get

```
get(url: str, **kwargs: Any) -> TestResponse
```

A GET request; see `request()`.

### head

```
head(url: str, **kwargs: Any) -> TestResponse
```

A HEAD request; see `request()`.

### options

```
options(url: str, **kwargs: Any) -> TestResponse
```

A OPTIONS request; see `request()`.

### post

```
post(url: str, **kwargs: Any) -> TestResponse
```

A POST request; see `request()`.

### put

```
put(url: str, **kwargs: Any) -> TestResponse
```

A PUT request; see `request()`.

### patch

```
patch(url: str, **kwargs: Any) -> TestResponse
```

A PATCH request; see `request()`.

### delete

```
delete(url: str, **kwargs: Any) -> TestResponse
```

A DELETE request; see `request()`.

### stream

```
stream(method: str, url: str, **kwargs: Any) -> _SyncStreamContext
```

Sends a request and returns the response before its body is read.

### websocket

```
websocket(url: str, **kwargs: Any) -> TestWebSocket
```

Opens a WebSocket, as a context manager; see `AsyncTestClient.websocket()`.

## AsyncTestClient

```
AsyncTestClient(app: Callable[..., Awaitable[None]], *, base_url: str = 'http://testserver', headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, raise_server_errors: bool = True, timeout: float | None = 5.0)
```

Calls an ASGI app in-process, from `async` tests.

As an async context manager it runs the app's lifespan (startup and shutdown hooks); without it, no hook runs.

- `base_url`: scheme and host requests go to, unless a request gives a full URL
- `headers`, `cookies`: sent with every request; a request's own replace the same name
- `raise_server_errors`: raise in the test an error the app logs as unhandled (a 500)
- `timeout`: seconds to wait for a response or WebSocket message, then `TimeoutError`; None waits forever

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

### request

```
request(method: str, url: str, *, query: Query | None = None, headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, json: Any = _UNSET, form: Form | None = None, files: Files | None = None, body: bytes | str | None = None, follow_redirects: bool = False) -> TestResponse
```

Sends a request and returns the whole response.

Give the body as `json`, `form` (with `files`: `{"f": ("a.png", data, "image/png")}`) or raw `body`, one way only (`TypeError` otherwise). `follow_redirects` follows up to 20 redirects (a 303, or a 301/302 after a POST, becomes a GET without body).

### stream

```
stream(method: str, url: str, *, query: Query | None = None, headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, json: Any = _UNSET, form: Form | None = None, files: Files | None = None, body: bytes | str | None = None) -> _StreamContext
```

Sends a request and returns the response before its body is read.

Leaving the block stops the app's handler, as a client going away would. Takes the arguments of `request()`, but not `follow_redirects`.

### get

```
get(url: str, **kwargs: Any) -> TestResponse
```

A GET request; takes the same arguments as `request()`.

### head

```
head(url: str, **kwargs: Any) -> TestResponse
```

A HEAD request; see `request()`.

### options

```
options(url: str, **kwargs: Any) -> TestResponse
```

A OPTIONS request; see `request()`.

### post

```
post(url: str, **kwargs: Any) -> TestResponse
```

A POST request; see `request()`.

### put

```
put(url: str, **kwargs: Any) -> TestResponse
```

A PUT request; see `request()`.

### patch

```
patch(url: str, **kwargs: Any) -> TestResponse
```

A PATCH request; see `request()`.

### delete

```
delete(url: str, **kwargs: Any) -> TestResponse
```

A DELETE request; see `request()`.

### websocket

```
websocket(url: str, *, subprotocols: Iterable[str] = (), query: Query | None = None, headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None) -> AsyncTestWebSocket
```

Opens a WebSocket, as an async context manager.

Entering raises `WebSocketDisconnect` if the app closes before accepting.

## TestResponse

```
TestResponse(status: int, headers: Headers, body: bytes, url: str)
```

A response: `status`, `headers`, `body`, `text`, `json()` and the final `url`.

### text

```
text: str
```

The body, decoded with the charset of its `Content-Type` (UTF-8 by default).

### json

```
json() -> Any
```

The body, parsed as JSON.

## TestWebSocket

```
TestWebSocket(websocket: AsyncTestWebSocket, run: Callable[[Awaitable[Any]], Any])
```

The sync `AsyncTestWebSocket`.

## AsyncTestWebSocket

```
AsyncTestWebSocket(client: AsyncTestClient, scope: dict)
```

A WebSocket open on the app: `subprotocol` and `headers` are those it accepted with.

### receive

```
receive(timeout: float | None = _UNSET) -> str | bytes
```

The next message from the app, text or bytes; `WebSocketDisconnect` once it closed.

`timeout` (seconds) overrides the client's.

### close

```
close(code: int = 1000) -> None
```

Disconnects (unless the app closed first) and waits for the handler to end.
# Optional

# Performance

On this page:

- [Servers](#servers): the built-in server against uvicorn, Granian, Robyn and FastrAPI
- [Under load](#under-load): frameworks on one CPU core
- [Router alone](#router-alone)
- [Validation](#validation) and [Dependencies](#dependencies)
- [Threads on free-threaded Python](#threads-on-free-threaded-python)
- [With a database](#with-a-database)
- [Memory under steady load](#memory-under-steady-load)
- [Templates](#templates)

All numbers come from the scripts in [`benches/`](https://github.com/3d3n-pyc/korvex/tree/main/benches), on one machine. Results depend on the machine: measure your own workload.

## Servers

The same app on the same 4 CPU cores, each server with its best number of workers. Measured with Korvex's main branch after 0.14.0 and FastrAPI 0.2.11:

| Server                              | Requests/s, 200 clients | 1,000 clients | RAM        |
| ----------------------------------- | ----------------------- | ------------- | ---------- |
| **Korvex built-in** (3 workers)     | **142.7k**              | **145.4k**    | **125 MB** |
| Granian + Korvex app (3 workers)    | 123.9k                  | 134.1k        | 198 MB     |
| uvicorn + Korvex app (4 workers)    | 117.5k                  | 110.4k        | 207 MB     |
| Robyn (4 processes, 2 threads each) | 95.7k                   | 91.7k         | 228 MB     |
| FastrAPI (1 process)                | 61.4k                   | 67.1k         | 113 MB     |

- Granian is a Rust ASGI server too, and comes second: the built-in server serves 8 to 15% more, because it skips the ASGI layer, with 40% less memory.
- Robyn and FastrAPI run their own version of the app on their own Rust server. FastrAPI runs one process (no workers option, and a second process can't share the port): its Rust side uses the 4 cores, but one Python process, under the GIL, runs every handler. Its handlers are sync: in 0.2.11, async ones answer a 500.
- No request failed.
- Per-request Python work between the server and the handler was cut (handler plan checked once per route, headers no longer checked one by one, no lookup before each Content-Type, no ASGI scope built and read back). One worker on one core serves 39.7k requests/s, 47% more than 0.14.0 (26.9k), measured side by side. On 4 cores the server's Rust threads and the kernel take a bigger share, so the gain is smaller.

Reproduce with `uv run --group bench python benches/servers.py` (needs `cargo install oha`, takes about 5 minutes).

## Under load

The same small app (1,000 routes, one JSON endpoint) in each framework, on uvicorn with one worker and one CPU core, flooded with requests to random URLs by [oha](https://github.com/hatoo/oha). Measured with Korvex's main branch after 0.14.0:

| Framework  | Requests/s, 200 clients | p99 latency, 1,000 clients | RAM at rest, peak |
| ---------- | ----------------------- | -------------------------- | ----------------- |
| **Korvex** | **34.4k**               | **24 ms**                  | **39 MB, 50 MB**  |
| Falcon     | 30.5k                   | 31 ms                      | 60 MB, 63 MB      |
| Starlette  | 3.0k to 30.3k           | 27 to 129 ms               | 39 MB, 49 MB      |
| Litestar   | 17.7k                   | 89 ms                      | 79 MB, 90 MB      |
| Sanic      | 14.1k                   | 99 ms                      | 61 MB, 105 MB     |
| FastAPI    | 1.6k to 12.6k           | 58 to 214 ms               | 66 MB, 77 MB      |
| BlackSheep | 7.0k                    | 68 ms                      | 51 MB, 59 MB      |
| Django     | 1.1k to 2.2k            | 1.2 s to 1.7 s             | 49 MB, 141 MB     |

- Every framework uses a full CPU core, so requests per second is the work each gets out of that core.
- Falcon, and Starlette in its best case, serve 11 to 12% less than Korvex.
- Starlette, FastAPI and Django try routes one by one: the ranges go from the tested route declared after the 1,000 others to declared first. BlackSheep also tries them one by one, but sorts them deepest first, so the tested route always comes last.
- No request failed. Throughput holds as clients are added; latency grows because requests queue. After the flood, memory stays flat and speed is back to normal.

Robyn and FastrAPI can't run on uvicorn, so they're compared with Korvex's built-in server above.

Reproduce with `uv run --group bench python benches/load_test.py` (needs `cargo install oha`, takes about 11 minutes).

## Router alone

Route matching only, without any server:

Median time per lookup, 10,000 routes:

| Router              | First time a URL is seen | Same URL again (cached) |
| ------------------- | ------------------------ | ----------------------- |
| **Korvex**          | **347 ns**               | **67 ns**               |
| Litestar            | 732 ns                   | 92 ns                   |
| Sanic               | 3.5 µs                   | 211 ns                  |
| Flask               | 3.5 µs                   | (no cache)              |
| Django              | 1.3 ms                   | (no cache)              |
| Starlette / FastAPI | 1.5 ms                   | (no cache)              |

- Korvex is the fastest in both cases, and its time stays flat as routes are added. Django and Starlette try routes one by one, so they slow down linearly.
- Korvex, Litestar and Sanic cache the last 1,024 lookups. It helps most with static URLs; URLs with parameters (`/users/1`, `/users/2`, ...) differ often, so they hit it less.
- With few routes, the server and handler cost more than the router. With many, a router that tries routes one by one becomes the main cost (see Under load).

Reproduce with `uv run --group bench pytest benches/bench_router.py --benchmark-only --benchmark-json=benches/results.json`, then `uv run --group bench python benches/plot.py` to redraw both charts.

## Validation

One POST with a small JSON body, answered with JSON, run through the app in-process (no server). Median time per request, Korvex's main branch after 0.14.0:

| Handler                                                      | Time        |
| ------------------------------------------------------------ | ----------- |
| **Korvex, msgspec models in and out**                        | **16.6 µs** |
| Korvex, no annotations (`request.json()`, `Response.json()`) | 19.8 µs     |
| Korvex, pydantic models in and out                           | 20.4 µs     |
| FastAPI, the same pydantic models                            | 73.5 µs     |

- With msgspec, validating is faster than not validating: it decodes and encodes JSON in compiled code, while `request.json()` and `Response.json()` use Python's `json` module. pydantic costs about the same as not validating.

Reproduce with `uv run --group bench pytest benches/bench_validation.py --benchmark-only`.

## Dependencies

One GET whose handler reads the user from a header and opens then closes a session, run in-process. Median time per request, Korvex's main branch after 0.14.0:

| Handler                                                        | Time        |
| -------------------------------------------------------------- | ----------- |
| Korvex, the handler does it itself                             | 16.2 µs     |
| **Korvex, two dependencies (one with `yield`) and `Header()`** | **23.7 µs** |
| FastAPI, the same dependencies, async                          | 93.8 µs     |
| FastAPI, the same dependencies, sync                           | 405 µs      |

- FastAPI runs sync dependencies in a thread pool: a blocking one can't stop the event loop, but it costs a lot. Korvex calls them on the loop, like sync handlers: keep blocking work out of them, or make them async.
- Dependencies are resolved when the route is registered; a handler without them pays nothing.

Reproduce with `uv run --group bench pytest benches/bench_dependencies.py --benchmark-only`.

## Threads on free-threaded Python

The built-in server on the same 4 cores, app and load (200 connections) as above, with more threads or workers. One run:

| Python, workers, threads        | Requests/s  | p99    | Memory     |
| ------------------------------- | ----------- | ------ | ---------- |
| 3.14t, 1 worker, 1 thread       | 71,096      | 4.6 ms | 39 MiB     |
| 3.14t, 1 worker, 2 threads      | 98,839      | 3.8 ms | 43 MiB     |
| **3.14t, 1 worker, 3 threads**  | **111,317** | 4.6 ms | **46 MiB** |
| 3.14t, 1 worker, 4 threads      | 103,378     | 6.0 ms | 49 MiB     |
| 3.14t, 2 workers, 1 thread      | 119,674     | 3.1 ms | 98 MiB     |
| 3.14t, 4 workers, 1 thread      | 116,183     | 4.1 ms | 159 MiB    |
| 3.14 (GIL), 1 worker, 1 thread  | 59,301      | 4.9 ms | 32 MiB     |
| 3.14 (GIL), 1 worker, 2 threads | 58,653      | 2.5 ms | 33 MiB     |
| 3.14 (GIL), 4 workers, 1 thread | 98,304      | 4.6 ms | 124 MiB    |

- On 3.14t, 3 threads serve 57% more than one, within 4% of 4 workers, with a third of the memory. Past 3 there is no gain: each loop has its own Rust thread, so 4 loops already keep 8 threads busy on 4 cores.
- With the GIL, threads don't help: use workers.
- Workers still serve the most; threads are for when memory counts (one copy of the app, caches and connection pools per process).
- `compress()` leaves brotli out on 3.14t: the `brotli` package turns the GIL back on for the whole process. CI checks that nothing else does.

Reproduce with `KORVEX_FT_PYTHON=/path/to/3.14t/python uv run --group bench python benches/threads.py` (see the file for the setup).

## With a database

Each request reads one row from PostgreSQL through an asyncpg pool and answers with JSON. PostgreSQL gets 2 cores, the server 4, with 3 workers each. Two runs:

| Server                  | Requests/s      | p99            | Memory  |
| ----------------------- | --------------- | -------------- | ------- |
| Korvex, built-in server | 21,660 / 21,380 | 37.7 / 32.4 ms | 133 MiB |
| Korvex on uvicorn       | 22,035 / 23,524 | 30.9 / 23.5 ms | 170 MiB |
| FastAPI on uvicorn      | 13,407 / 15,356 | 41.7 / 41.6 ms | 229 MiB |

- The database sets the pace: Korvex's two servers serve about the same (uvicorn a little more in the second run), and the built-in server's lead from the tests above is gone. It still uses the least memory.
- FastAPI serves 35 to 40% less: its own cost per request adds to the database's.

Reproduce with `uv run --group bench python benches/database.py` (needs podman or docker for PostgreSQL, and oha).

## Memory under steady load

`benches/soak.py` runs the built-in server with 2 workers for 10 minutes of mixed load: keep-alive and new connections, JSON POST, streams, files, WebSocket echo, slow clients and clients that hang up mid-stream. It samples each worker's memory every 5 seconds. It fails if a worker grows over 10% and 4 MiB after warm-up, on any error, or if a worker restarts. A nightly CI run uses it. Over 10 minutes (900,000 HTTP requests, 3.8 million WebSocket messages, 870,000 streams cut off), the workers grew +0.7% and +0.1%, at 27 MiB each. With a deliberate leak (`SOAK_LEAK=1`), they grew +150% and the test failed.

Run it with `uv run python benches/soak.py --duration 60` (Linux, needs `oha`).

## Templates

Rendering a template with autoescape on, precompiled and cached, against Jinja2 and minijinja (a Rust engine). Median time per render, in µs, on one machine:

| Template                              | Korvex   | Jinja2 | minijinja |
| ------------------------------------- | -------- | ------ | --------- |
| A table of 100 rows                   | **88.5** | 444    | 86        |
| A layout with `extends` and `include` | **9.7**  | 59     | 11.7      |
| An escape-heavy form                  | **43**   | 190    | 40        |

- Korvex is 4 to 6 times faster than Jinja2, and on par with minijinja.
- The outputs are checked equal to Jinja2's before timing.
- The time is spent in the template: the app, the server and the request are not part of it.

Reproduce with `uv run --group bench python benches/bench_templates.py`.

# The standalone router

`korvex.router` is Korvex's Rust router on its own, to build something of your own on top. It maps a path and a method to a handler **name** and never calls your code. An app doesn't need it: `App` uses it inside.

```
from korvex.router import Router

router = Router()
router.get("/users/{id:int}", "get_user")
router.post("/users", "create_user", middleware=["auth"])

router.match("/users/42")  # ("get_user", {"id": 42}, ())
router.match("/users", "POST")  # ("create_user", {}, ("auth",))
router.match("/nope")  # None
router.match("/users", "DELETE")  # raises MethodNotAllowed, .allowed == ["POST"]
```

The path syntax is the App's: see [Routes](https://3d3n-pyc.github.io/korvex/routes/#path-parameters).

| Method                                                     | Does                                                                   |
| ---------------------------------------------------------- | ---------------------------------------------------------------------- |
| `route(path, handler, *, method="GET", middleware=None)`   | Registers a route, or replaces it.                                     |
| `get`, `post`, `put`, `patch`, `delete`, `head`, `options` | Shortcuts for `route`.                                                 |
| `match(path, method="GET")`                                | `(handler, params, middleware)`, `None`, or raises `MethodNotAllowed`. |
| `remove(path, method="GET")`                               | Removes a route; returns whether it was there.                         |
| `routes()`                                                 | Every route, as `(path, method, handler, middleware)`.                 |
| `group(prefix, *, middleware=None)`                        | A `RouteGroup`, below.                                                 |
| `constraint(name, regex)`, `constraints()`                 | Your own parameter types: `{post:slug}`.                               |

```
api = router.group("/api", middleware=["auth"])
api.get("/users", "list_users")  # GET /api/users, middleware ("auth",)
```

A group has `get()`, `post()`... like the router, `group()` for a nested one, and `full_path("users")`, which gives `/api/users`.

- `middleware` names are given back by `match()`, for your code to run.
- `match()` caches the last 1,024 lookups; adding or removing a route clears the cache. Results are read-only, as they're shared.
- An invalid path or method raises `ValueError`.

# For AI assistants

Korvex is young: AI assistants know little about it, and may know it wrong. Give them its documentation:

## The documentation, for LLMs

The site also publishes Markdown, rebuilt with every change:

- [`llms.txt`](https://3d3n-pyc.github.io/korvex/llms.txt): an index of every page, with a line on each.
- [`llms-full.txt`](https://3d3n-pyc.github.io/korvex/llms-full.txt): the whole guide and reference in one file, to paste into a chat or give to a tool that reads URLs.
- Every page also exists as Markdown: add `index.md` to its URL, like <https://3d3n-pyc.github.io/korvex/routes/index.md>.

## A skill for coding agents

The repository has an [agent skill](https://github.com/3d3n-pyc/korvex/tree/main/.agents/skills/korvex): a short `SKILL.md` that tells an agent how to write a Korvex app, with a complete example and the rules that differ from FastAPI. Korvex's test suite runs its examples.

It comes with Korvex. In your project's folder, run once:

```
korvex agents
```

It writes the skill, in the installed Korvex's version, to `.agents/skills/korvex/` (Codex, Cursor and most agents) and `.claude/skills/korvex/` (Claude Code). Rerun it after upgrading Korvex; commit both folders.

Without Python at hand, the [skills CLI](https://github.com/vercel-labs/skills) installs the latest one from GitHub: `npx skills add 3d3n-pyc/korvex`.

An agent that opens `korvex/__init__.py`, or runs `help(korvex)`, also finds a note pointing it to `llms-full.txt` and `korvex agents`.

## Working on Korvex itself

[`AGENTS.md`](https://github.com/3d3n-pyc/korvex/blob/main/AGENTS.md), at the root of the repository, gives coding agents the checks to run and the rules to follow. Most agents read it on their own.