Types
Core request/response types and state containers.
Request & Response
Request
class Request(method: str, path: str, raw_query_string: bytes = b'', headers: Headers | None = None, path_params: dict[str, str] | None = None, client: ClientInfo | None = None, server: ServerInfo | None = None, receive: Callable[[], Awaitable[dict[str, Any]]] | None = None, app_state: AppState | None = None, max_body_size: int = 1048576)Incoming HTTP request.
The request owns its ASGI scope and the receive callable required to
consume the body. State, route metadata, and app state are attached by the
runtime before the handler executes.
Request.reset
def reset(self, method: str, path: str, raw_query_string: bytes, headers: Headers, client: ClientInfo, server: ServerInfo, receive: Callable[[], Awaitable[dict[str, Any]]], app_state: AppState, max_body_size: int) -> NoneRe-initialise this •Request in place for reuse.
The •lauren._arena.RequestArena pools Request
instances along with its container dicts. reset() lets the
dispatcher hand the same object to a new request without
re-running __init__ — the saving is small per-call but
compounds measurably under load.
Every attribute set by __init__ is re-set here; per-request
caches (_query_params, _cookies, _body) are cleared
so the previous request's data cannot leak across the pool.
The route-metadata slots (_matched_route etc.) are wiped
too — the dispatcher re-populates them after routing.
Request.body
def body(self) -> bytesRequest.text
def text(self, encoding: str = 'utf-8') -> strRequest.json
def json(self) -> AnyRequest.form
def form(self) -> dict[str, list[str]]Request.stream
def stream(self) -> AsyncIterator[bytes]Request.get_handler_class
def get_handler_class(self) -> type | NoneRequest.get_route_handler_func
def get_route_handler_func(self) -> Callable[..., Any] | NoneRequest.get_route_template
def get_route_template(self) -> str | NoneRequest.get_matched_route
def get_matched_route(self) -> Any | NoneResponse
class Response(body: bytes | str | None = b'', status: int = 200, headers: Headers | MutableHeaders | None = None, media_type: str | None = None, stream: AsyncIterable[bytes] | None = None)Immutable HTTP response value object.
Mutating methods (with_*) return a new instance. Bodies may be a
bytes blob or an async iterable for streaming responses.
Response.json
def json(cls, data: Any, status: int = 200, headers: Headers | None = None, encoder: Any = None) -> 'Response'Build a JSON response.
When encoder is provided it must implement the
•lauren.serialization.JSONEncoder protocol — the
dispatcher passes in the app's active encoder so every
response uses the configured backend. When omitted (e.g.
tests that build responses directly, or pre-app call sites),
falls back to the process-wide default which starts as the
stdlib encoder and can be swapped via
•lauren.serialization.set_active_encoder.
Response.text
def text(cls, data: str, status: int = 200, headers: Headers | None = None) -> 'Response'Response.html
def html(cls, data: str, status: int = 200, headers: Headers | None = None) -> 'Response'Response.bytes
def bytes(cls, data: bytes, status: int = 200, media_type: str = 'application/octet-stream', headers: Headers | None = None) -> 'Response'Response.empty
def empty(cls, status: int = 204) -> 'Response'Response.no_content
def no_content(cls) -> 'Response'Response.created
def created(cls, data: Any | None = None, location: str | None = None) -> 'Response'Response.accepted
def accepted(cls, data: Any | None = None) -> 'Response'Response.redirect
def redirect(cls, location: str, status: int = 307) -> 'Response'Response.stream
def stream(cls, iterable: AsyncIterable[bytes], status: int = 200, media_type: str = 'application/octet-stream', headers: Headers | None = None) -> 'Response'Response.sse
def sse(cls, iterable: AsyncIterable[str | dict[str, Any]], status: int = 200) -> 'Response'Response.with_status
def with_status(self, status: int) -> 'Response'Response.with_header
def with_header(self, key: str, value: str) -> 'Response'Response.with_headers
def with_headers(self, mapping: Mapping[str, str]) -> 'Response'Response.without_header
def without_header(self, key: str) -> 'Response'Response.with_media_type
def with_media_type(self, media_type: str) -> 'Response'Response.with_body
def with_body(self, body: bytes | str) -> 'Response'Response.with_cookie
def with_cookie(self, key: str, value: str, max_age: int | None = None, path: str = '/', domain: str | None = None, secure: bool = False, http_only: bool = False, same_site: str | None = None) -> 'Response'Response.delete_cookie
def delete_cookie(self, key: str, path: str = '/') -> 'Response'Headers
class Headers(items: list[tuple[str, str]] | None = None)Case-insensitive, ordered, multi-value header container.
The primary lookup returns the first value for simplicity; use
•getall to retrieve every value for a header name.
Headers.get
def get(self, key: str, default: Any = None) -> AnyHeaders.getall
def getall(self, key: str) -> list[str]Headers.raw
def raw(self) -> list[tuple[str, str]]Headers.mutable_copy
def mutable_copy(self) -> 'MutableHeaders'MutableHeaders
class MutableHeadersMutable variant used when building responses.
MutableHeaders.set
def set(self, key: str, value: str) -> NoneMutableHeaders.append
def append(self, key: str, value: str) -> NoneMutableHeaders.delete
def delete(self, key: str) -> NoneClientInfo
class ClientInfo(host: str | None, port: int | None)State
State
class State(initial: Mapping[str, Any] | None = None)Request-scoped state bag with typed accessors.
Attributes can be set either via attribute-style (state.user = u) or
with •set. Typed retrieval helps middleware/handlers avoid silent
type errors.
State.set
def set(self, key: str, value: Any) -> NoneState.get
def get(self, key: str, default: Any = None) -> AnyState.has
def has(self, key: str) -> boolState.get_typed
def get_typed(self, key: str, expected: type[T]) -> T | NoneState.require
def require(self, key: str, expected: type[T]) -> TState.asdict
def asdict(self) -> dict[str, Any]AppState
class AppState(initial: Mapping[str, Any] | None = None)Read-only application-level state.
Writes raise •RuntimeError after the app has been sealed.
AppState.seal
def seal(self) -> NoneAppState.set
def set(self, key: str, value: Any) -> None