Skip to content

Requests and responses

Request

Immutable view of an incoming HTTP request, handed to the Python handler. frozen means no Python-side mutation, so no locking is needed even on free-threaded builds.

app property

app

The app serving this request.

body property

body

The raw request body. A pydantic-annotated argument is the usual way to read a body; this is for handlers that parse it themselves.

cookies property

cookies

Cookies parsed from the Cookie header.

headers property

headers

Every header, lowercased. Repeated headers are joined with ", " as HTTP itself defines.

method property

method

The HTTP method, uppercase.

path property

path

The request path, without the query string.

query property

query

The raw query string, or None. Declared query parameters are already coerced and passed as handler arguments; this is for the rest.

state property

state

Values yielded by the app's lifespan and worker_lifespan, read-only.

form method descriptor

form(max_parts=1000)

Parse the body as a form, application/x-www-form-urlencoded or multipart/form-data, into a FormData.

Parsed each time it is called, and only when it is called. Raises HTTPError(415) for a body that is not a form, HTTPError(400) for a malformed one and HTTPError(413) past max_parts, which bounds how many Python objects a single request can make the worker build.

header method descriptor

header(name, default=None)

One header by name, case-insensitively. None if absent.

This is the cheap path: no dict is built, and a header that is not valid UTF-8 reads as absent rather than raising.

stream method descriptor

stream()

The body as an async iterator of bytes chunks: a BodyStream.

On a route that declares a BodyStream argument the chunks arrive as the client sends them, and nothing is read until the first one is asked for. On any other route the body was already collected, and this yields it as a single chunk, so code reading a stream works on both.

Response dataclass

Response(
    body: bytes | str = b"",
    status: int = 200,
    content_type: str = "application/json",
    headers: dict[str, str] = dict(),
)

A ready-to-send response.

body may be bytes or str; str is encoded as UTF-8. headers are sent in addition to the content type.

body class-attribute instance-attribute

body: bytes | str = b''

status class-attribute instance-attribute

status: int = 200

content_type class-attribute instance-attribute

content_type: str = 'application/json'

headers class-attribute instance-attribute

headers: dict[str, str] = field(default_factory=dict)

encoded

encoded() -> bytes
Source code in python/oxbrook/_response.py
def encoded(self) -> bytes:
    return self.body.encode() if isinstance(self.body, str) else bytes(self.body)

header_list

header_list() -> list[tuple[str, str]] | None
Source code in python/oxbrook/_response.py
def header_list(self) -> list[tuple[str, str]] | None:
    return list(self.headers.items()) if self.headers else None

Reply

Reply(
    value: Any = None,
    status: int | None = None,
    headers: dict[str, str] | None = None,
)

A handler's result on its way back out.

value is whatever the handler returned: a dict, a model, a Response, an SSE, or None. status overrides what that value would otherwise imply. headers are added to the response.

Source code in python/oxbrook/_middleware.py
def __init__(
    self,
    value: Any = None,
    status: int | None = None,
    headers: dict[str, str] | None = None,
) -> None:
    self.value = value
    self.status = status
    self.headers = headers if headers is not None else {}

value instance-attribute

value = value

status instance-attribute

status = status

headers instance-attribute

headers = headers if headers is not None else {}

FormData

FormData(items: list[tuple[str, Any]])

Bases: Mapping

Parsed form fields, in the order they arrived.

Indexing returns the first value for a name, as most forms have one; use getlist for a field that repeats. Text fields are str, files are UploadFile.

Source code in python/oxbrook/_forms.py
def __init__(self, items: list[tuple[str, Any]]) -> None:
    self._items = items

files property

files: list[UploadFile]

from_parts classmethod

from_parts(parts: list[tuple]) -> FormData
Source code in python/oxbrook/_forms.py
@classmethod
def from_parts(cls, parts: list[tuple]) -> "FormData":
    items: list[tuple[str, Any]] = []
    for part in parts:
        if len(part) == 2:
            items.append((part[0], part[1]))
        else:
            name, filename, content_type, data = part
            items.append((name, UploadFile(name, filename, content_type, data)))
    return cls(items)

getlist

getlist(name: str) -> list[Any]
Source code in python/oxbrook/_forms.py
def getlist(self, name: str) -> list[Any]:
    return [value for key, value in self._items if key == name]

multi_items

multi_items() -> list[tuple[str, Any]]

Every (name, value) pair, repeats included.

Source code in python/oxbrook/_forms.py
def multi_items(self) -> list[tuple[str, Any]]:
    """Every (name, value) pair, repeats included."""
    return list(self._items)

UploadFile

UploadFile(
    name: str,
    filename: str,
    content_type: str | None,
    data: bytes,
)

One file from a multipart form.

Source code in python/oxbrook/_forms.py
def __init__(self, name: str, filename: str, content_type: str | None, data: bytes) -> None:
    self.name = name
    #: As the client sent it. Never use it as a filesystem path unchecked:
    #: it is client input, and may be empty or contain `..` and slashes.
    self.filename = filename
    self.content_type = content_type
    self.data = data

name instance-attribute

name = name

filename instance-attribute

filename = filename

content_type instance-attribute

content_type = content_type

data instance-attribute

data = data

size property

size: int

read

read() -> bytes
Source code in python/oxbrook/_forms.py
def read(self) -> bytes:
    return self.data

text

text(encoding: str = 'utf-8') -> str
Source code in python/oxbrook/_forms.py
def text(self, encoding: str = "utf-8") -> str:
    return self.data.decode(encoding)

Form

Form(*, max_parts: int = 1000)

Marks a pydantic-model argument as bound from a form body.

async def signup(_: Request, data: Signup = Form()): ...

max_parts bounds how many fields and files one request may send.

Source code in python/oxbrook/_forms.py
def __init__(self, *, max_parts: int = 1000) -> None:
    if max_parts < 1:
        raise ValueError("max_parts must be at least 1")
    self.max_parts = max_parts

max_parts instance-attribute

max_parts = max_parts

BodyStream

BodyStream(reader: Any, buffered: bytes = b'')

The request body, as an async iterator of bytes.

Use as a handler argument's annotation to make a route stream, or call request.stream(). On a route that did not stream, it yields the already-collected body once.

Source code in python/oxbrook/_bodies.py
def __init__(self, reader: Any, buffered: bytes = b"") -> None:
    self._reader = reader
    self._buffered = buffered

read async

read() -> bytes

The whole remaining body. Bounded by max_body, like any body.

Source code in python/oxbrook/_bodies.py
async def read(self) -> bytes:
    """The whole remaining body. Bounded by `max_body`, like any body."""
    return b"".join([chunk async for chunk in self])

CORS dataclass

CORS(
    allow_origins: Iterable[str],
    allow_methods: Iterable[str] = ("*",),
    allow_headers: Iterable[str] = ("*",),
    allow_credentials: bool = False,
    expose_headers: Iterable[str] = tuple(),
    max_age: int | None = 600,
)

Which other origins may call this app from a browser.

allow_origins lists exact origins — scheme, host and port as the browser sends them — or ["*"] for any.

allow_credentials lets the browser send cookies and Authorization on those requests. It cannot be combined with "*": that would let every website make authenticated requests as a signed-in user, so it raises here rather than being quietly made to work.

expose_headers lists response headers a page's script may read, beyond the few a browser always exposes. max_age is how long, in seconds, a browser may cache a preflight answer.

allow_origins instance-attribute

allow_origins: Iterable[str]

allow_methods class-attribute instance-attribute

allow_methods: Iterable[str] = ('*',)

allow_headers class-attribute instance-attribute

allow_headers: Iterable[str] = ('*',)

allow_credentials class-attribute instance-attribute

allow_credentials: bool = False

expose_headers class-attribute instance-attribute

expose_headers: Iterable[str] = field(default_factory=tuple)

max_age class-attribute instance-attribute

max_age: int | None = 600

as_spec

as_spec() -> tuple

The tuple the Rust server takes.

Source code in python/oxbrook/_cors.py
def as_spec(self) -> tuple:
    """The tuple the Rust server takes."""
    return (
        list(self.allow_origins),
        list(self.allow_methods),
        list(self.allow_headers),
        self.allow_credentials,
        list(self.expose_headers),
        self.max_age,
    )