fastware v0.6.0 /src.fastware.request
Updated
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).

Search