Skip to content
src.fastware.request
Updated Edit
On this page

Request wrapper with lazy msgspec JSON parsing, json_as() decoding, typed query-parameter extraction, header/cookie access, and State object.

#src.fastware.request

#src.fastware.request

HTTP request wrapper providing lazy body parsing, query parameter extraction, JSON deserialization via msgspec, header access, and per-request state.

#State

Dict-backed state that supports both attribute and dict access.

state.key, state["key"], and state.get("key") all work. Attribute assignment (state.key = val) also works.

#get

python
def get(self, key: str, default: Any=None) -> Any

#Request

Wraps ASGI scope with parsed body and params.

#json

python
def json(self) -> dict | list | None

Lazily decode the JSON body on first access, then cache.

#body

python
def body(self) -> bytes | None

Raw request body bytes.

#_qs

python
def _qs(self) -> dict[str, list[str]]

Full parse_qs of the query string (name -> list of values).

Parsed once on first access and cached, then shared by query(), query_list(), and query_params so the query string is never re-parsed per call.

#query

python
def query(self, name: str, default: Any=_MISSING, *, type_: type=str, ge: int | float | None=None, le: int | float | None=None, min_length: int | None=None, max_length: int | None=None) -> Any

Get a query parameter by name, with optional type conversion and constraints.

The default is only used when the key is absent from the query string. When no default is given and the key is absent, returns None.

Type coercion failure (key present but unconvertible) always raises HTTPError(422) -- the default is not used as a fallback for bad input.

Constraints (checked after type coercion):

  • ge: value must be >= this (numeric)
  • le: value must be <= this (numeric)
  • min_length: len(value) must be >= this (strings)
  • max_length: len(value) must be <= this (strings)

Raises HTTPError(422) on coercion failure or constraint violation.

#query_list

python
def query_list(self, name: str, type_: type=str) -> list

Return all values for a multi-value query key with optional type coercion.

Returns an empty list if the key is absent. Raises HTTPError(422) if any value cannot be converted to type_.

#query_params

python
def query_params(self) -> dict[str, str]

Parsed query string as a dict (first value per key). Cached.

python
def header(self, name: str, default: str | None=None) -> str | None

Return a request header by name (case-insensitive).

ASGI headers arrive as a list of (bytes, bytes) tuples. This decodes to str and looks the name up case-insensitively, matching HTTP semantics.

#body_size

python
def body_size(self) -> int

Length in bytes of the raw request body (0 if no body).

#state

python
def state(self) -> State

Lifespan + per-request state, supporting both attribute and dict access. Cached.

#method

python
def method(self) -> str

HTTP method (GET, POST, etc.).

#path

python
def path(self) -> str

Request path.

#is_disconnected

python
async def is_disconnected(self) -> bool

Return whether the client has disconnected.

This is a pure read of a flag maintained by the app's disconnect watcher, which is the single owner of the ASGI receive channel for the request's post-body lifetime. Because it never touches receive (no channel peeking, no receive task), it is safe to call any number of times from both streaming and non-streaming handlers and never competes with the watcher for the single receive channel.

#_mark_disconnected

python
def _mark_disconnected(self) -> None

Record a client disconnect. Called by the app's disconnect watcher.

Sets the flag and cancels any in-flight stream-driving task so a generator blocked between yields is unwound promptly (running its finally cleanup) instead of leaking until process exit.

#_register_stream_task

python
def _register_stream_task(self, task: Any) -> None

Register the task driving a streaming response body so the watcher can cancel it on disconnect.

#cookies

python
def cookies(self) -> dict[str, str]

Parse Cookie header and return a dict of cookie name-value pairs.

python
def cookie(self, name: str, default: str | None=None) -> str | None

Get a single cookie value by name.

#json_as

python
def json_as(self, model: type)

Parse the request body into model, dispatching on the target type.

If model is a msgspec.Struct subclass, the raw body is decoded with msgspec.json.decode(body, type=model) -- the fast, msgspec-native path the framework is built around. Otherwise the body falls back to the Pydantic path (model.model_validate). Either way, a decode/validation failure raises HTTPError(422).

More tools from this site

  • claudestream Drive Claude Code from Python: run it as a subprocess and read its output as typed events, with async and sync sessions, sandbox policies, and tools you define in Python
  • claudewheel A TUI Claude Code Launcher that lets you have more than one profile, manage sessions lifecycle, pick the exact CC version, model to use (even older unlisted ones), pick which GitHub account to use, etc.
  • dirstat Fast, single-binary directory statistics CLI: every file under a tree grouped by format, with counts, sizes, and lines of code, as a colored terminal table or as JSON
  • go-toml-edit Zero-dep TOML editing library for Go with comment preservation
  • howmuchleft The fastest Claude Code statusline: context window, 5-hour, and weekly limit usage as three customizable gradient bars, rendering in about 6 ms
  • orxtra
  • pgdesign
  • predraw Declarative rendering pipeline: describe a scene in JSON and get SVG, PNG and WebP out, with light and dark style tokens, reusable components and text converted to path outlines
  • reposummary Turn a git repository's history into a Markdown journal: pick a time window or revision range and get a readable digest of what changed, optionally narrated by an LLM
  • rlsbl Release orchestration and project scaffolding CLI that bumps versions, validates a structured JSONL changelog, tags only the commit CI verified, and publishes to npm, PyPI, Go and more
  • safegit git wrapper CLI that gives each commit its own temporary index and retries ref updates on conflict, so concurrent agents share one repository
  • saferm Command-line replacement for rm that archives every deletion with a mandatory reason and the context it ran in, so deleted files can be listed, inspected and restored
  • selfdoc Static Site Generator that builds a project's documentation site directly from its source code, so the docs can never drift from the code they describe, with SEO/AEO, first-class blog, search, and cross-project linking built in
  • strictcli
  • stricttest An always-on test-isolation floor: a pytest plugin and a Go env-hygiene module that make a test suite structurally unable to reach real credentials, the real HOME, the network, or the development repository.
  • wesktop A Python framework that turns an ASGI web app into a desktop application, serving it from a local Granian server and displaying it in a native OS window via pywebview
Search