#Core API Reference

This page documents fastware's core modules: the foundational types, response classes, request handling, routing, WebSocket support, and the application factory. These are the building blocks for every fastware application.

#ASGI Types

Low-level ASGI type aliases (Scope, Receive, Send) used throughout fastware. These type aliases are re-exported from the top-level package for convenience and provide consistent type-checking across the entire middleware and routing stack.

#src.fastware.types

ASGI type aliases (Scope, Receive, Send) used throughout fastware for consistent type-checked request and response handling.

#Scope

Scope = dict[str, Any]

#Receive

Receive = Callable[[], Awaitable[dict[str, Any]]]

#Send

Send = Callable[[dict[str, Any]], Awaitable[None]]

#Response Types

HTTP response types for returning data from route handlers. Handlers can return any of these types, or plain dict/list values (which are automatically wrapped in JSONResponse). Also includes cookie helpers and the HTTPError exception for error responses.

#src.fastware.responses

HTTP response types including JSON, text, HTML, bytes, and streaming responses, plus cookie helpers and low-level ASGI send functions.

#set_cookie

def set_cookie(name: str, value: str, *, httponly: bool=False, samesite: str='lax', max_age: int | None=None, path: str='/', secure: bool=False) -> str

Build a Set-Cookie header string.

#delete_cookie

def delete_cookie(name: str, *, path: str='/') -> str

Build a Set-Cookie header string that clears the cookie.

#HTTPError

Raise from handlers to return a specific HTTP error status.

#JSONResponse

JSON response with optional status code, headers, and cookies.

#TextResponse

Plain text or CSS response.

headers (optional) is merged with the framework's default response headers; values must already be plain strings.

#HTMLResponse

HTML response.

#BytesResponse

Raw bytes response with an explicit content type.

#StreamResponse

Streaming response (for SSE).

#FileResponse

Serve a file from disk with MIME detection and Content-Length.

#_build_headers

def _build_headers(content_type: str, extra_headers: dict[str, str] | None=None, cookies: list[str] | None=None) -> list[list[bytes]]

Assemble the ASGI response header list from content type, extra headers, and Set-Cookie values.

Returns headers in the order: content-type, then each extra header, then each cookie as a separate set-cookie entry. Callers that know the body length (buffered responses) insert content-length after content-type; streaming responses omit it.

#_send_response

async def _send_response(send: Callable, status: int, body: bytes, content_type: str, extra_headers: dict[str, str] | None=None, cookies: list[str] | None=None) -> None

Send a complete HTTP response (headers + body).

#send_error

async def send_error(send: Callable, status: int, detail: str) -> None

Send a complete JSON error response via raw ASGI send calls.

Intended for use by middleware that needs to short-circuit with an error without constructing response objects.

#Request Handling

The Request wrapper provides lazy body parsing, query parameter extraction with type coercion and validation, header access, cookie parsing, and per-request state. Handlers receive a Request as their first argument.

#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

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

#Request

Wraps ASGI scope with parsed body and params.

#json

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

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

#body

def body(self) -> bytes | None

Raw request body bytes.

#_qs

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

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

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

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

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

#header

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

def body_size(self) -> int

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

#state

def state(self) -> State

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

#method

def method(self) -> str

HTTP method (GET, POST, etc.).

#path

def path(self) -> str

Request path.

#is_disconnected

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

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

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

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

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

#cookie

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

Get a single cookie value by name.

#json_as

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).

#Routing

Path-based HTTP router with {param} placeholder syntax, typed parameters ({id:int}), greedy path segments ({path:path}), and sub-router composition via include_router. Supports all 7 standard HTTP methods (GET, POST, PUT, PATCH, DELETE, OPTIONS, HEAD) and WebSocket routes.

#src.fastware.routing

Path-based HTTP router with curly-brace parameter placeholders, automatic type coercion, method-based dispatch, and route group composition.

#_parse_segment

def _parse_segment(seg: str) -> tuple[str | None, str | None, type | None]

Parse a route pattern segment into (literal, param_name, converter).

Returns one of:

• (literal_str, None, None) for plain segments like "api"
• (None, param_name, converter) for parameterized segments like {id:int}
• (None, param_name, None) for :path segments (greedy)

#Router

Simple path-based HTTP router using {param} placeholders and type coercion.

Supports {param}, {param:str}, {param:int}, and {param:path} syntax.

#mount

def mount(self, prefix: str, app: Any) -> None

Mount an ASGI sub-application at a path prefix.

When a request path starts with prefix, the scope is rewritten (path stripped, root_path extended) and forwarded to app. Both http and websocket scope types are forwarded.

The prefix must start with / and must not end with /. A trailing slash is stripped automatically.

#_method_decorator

def _method_decorator(self, method: str, path: str, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> Callable

Return a decorator that registers a handler for method at path.

Shared factory backing the get/post/put/patch/delete decorators.

#get

def get(self, path: str, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> Callable

Decorator to register a GET handler.

#post

def post(self, path: str, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> Callable

Decorator to register a POST handler.

#delete

def delete(self, path: str, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> Callable

Decorator to register a DELETE handler.

#put

def put(self, path: str, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> Callable

Decorator to register a PUT handler.

#patch

def patch(self, path: str, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> Callable

Decorator to register a PATCH handler.

#add_route

def add_route(self, method: str, path: str, handler: Callable, *, deps: dict[str, Callable] | None=None, response_model: type | None=None) -> None

Programmatic route registration.

#ws

def ws(self, path: str, *, deps: dict[str, Callable] | None=None) -> Callable

Decorator to register a WebSocket handler.

#add_ws_route

def add_ws_route(self, path: str, handler: Callable, *, deps: dict[str, Callable] | None=None) -> None

Register a WebSocket handler for a path pattern (supports {param}).

#include_router

def include_router(self, other: Router, prefix: str | None=None, deps: dict[str, Callable] | None=None) -> None

Copy all routes from other into this router.

If prefix is given (e.g. "/api/v1"), its segments are prepended to every copied route's pattern.

If deps is given (a dict mapping names to factory callables), they are merged into each copied route's deps. Router-level deps are listed first so that per-handler deps can override them.

#match

def match(self, method: str, path: str) -> tuple[Callable, dict[str, Any]] | None

Return (handler, path_params) or None if no route matches.

Path parameter values are coerced to their declared types (e.g., {id:int} produces an int). If coercion fails the route does not match, allowing fall-through to 404.

#_match_pattern

def _match_pattern(cls, pattern: list[ParsedSegment], segments: list[str]) -> dict[str, Any] | None

Return coerced path params if pattern matches segments, else None.

Handles both greedy {param:path} patterns and normal segment-count patterns. Shared by HTTP and method-agnostic matching.

#_match_with_deps

def _match_with_deps(self, method: str, path: str) -> tuple[Callable, dict[str, Any], dict[str, Callable], type | None] | None

Return (handler, path_params, deps, response_model) or None.

Internal variant of :meth:match that also returns the merged dependency dict and response_model for the matched route. Used by create_app for DI resolution and response validation.

A HEAD request with no explicit HEAD route falls back to the matching GET route (the caller is responsible for sending an empty body). Callers wanting to distinguish a missing path (404) from a method mismatch (405) can consult :meth:allowed_methods.

#allowed_methods

def allowed_methods(self, path: str) -> set[str]

Return the set of HTTP methods registered for routes matching path.

Used to distinguish a missing path (empty set -> 404) from a method mismatch (non-empty set -> 405). When GET is registered for a matching route, HEAD is included as well, since HEAD is served by the GET handler.

#_match_with_path_param

def _match_with_path_param(pattern: list[ParsedSegment], segments: list[str], path_idx: int) -> dict[str, Any] | None

Match a route pattern containing a :path greedy parameter.

Literal/typed segments before the :path param must match exactly. Literal/typed segments after the :path param are matched from the end of the path. Everything in between is consumed by the :path parameter (joined with "/").

#match_ws

def match_ws(self, path: str) -> tuple[Callable, dict[str, Any]] | None

Return (handler, path_params) for a WebSocket path, or None.

#_match_ws_with_deps

def _match_ws_with_deps(self, path: str) -> tuple[Callable, dict[str, Any], dict[str, Callable]] | None

Return (handler, path_params, deps) for a WebSocket path, or None.

Internal variant of :meth:match_ws that also returns deps.

#WebSocket

WebSocket helper class wrapping the raw ASGI scope/receive/send triple into a convenient interface with accept, close, send_json, receive_json, send_text, receive_text, and similar methods. Handlers registered via router.ws() receive a WebSocket instance instead of a raw Request.

#src.fastware.websocket

WebSocket helper class wrapping the raw ASGI scope/receive/send triple with typed accept, send, receive, and close methods for ergonomic usage.

#WebSocketDisconnect

Raised when a WebSocket client disconnects.

The code attribute carries the close code from the ASGI websocket.disconnect message (defaults to 1000 / normal closure).

#WebSocket

Wraps the raw ASGI (scope, receive, send) triple for WebSocket connections.

Handlers receive a WebSocket instance instead of the raw triple, providing convenient methods for accept/close/send/receive and properties for path_params, headers, and query_string.

#path_params

def path_params(self) -> dict[str, Any]

#headers

def headers(self) -> dict[str, str]

Parse ASGI headers into a case-preserving dict (first value wins).

#query_string

def query_string(self) -> str

#accept

async def accept(self, subprotocol: str | None=None) -> None

#close

async def close(self, code: int=1000) -> None

#send_json

async def send_json(self, data: Any) -> None

#send_bytes

async def send_bytes(self, data: bytes) -> None

#send_text

async def send_text(self, text: str) -> None

#_receive_data

async def _receive_data(self) -> dict[str, Any]

Receive a data message, raising WebSocketDisconnect on disconnect.

#receive_json

async def receive_json(self) -> Any

#receive_bytes

async def receive_bytes(self) -> bytes

#receive_text

async def receive_text(self) -> str

#receive_raw

async def receive_raw(self) -> dict[str, Any]

Return the raw ASGI message dict from the WebSocket connection.

The dict contains keys like "type", "bytes", "text" depending on the frame type. Useful for handlers that need to distinguish between binary and text frames without committing to one receive method.

#Application Factory

The create_app function assembles a Router, optional middleware, static file serving, SPA fallback, lifespan management, and built-in middleware (CORS, request ID, request timing, trusted hosts, Vite dev proxy) into a single ASGI application callable.

#src.fastware.app

ASGI application factory with middleware chain composition, static file serving, SPA fallback routing, async lifespan hooks, and WebSocket support.

#_disconnect_watcher

async def _disconnect_watcher(receive: Callable, request: Request) -> None

Own the ASGI receive channel for a request's post-body lifetime.

A single watcher task is the only consumer of receive once the body has been read. It blocks on receive() and, on http.disconnect, records the fact on the request (which also cancels any in-flight stream-driving task) and returns. This makes disconnects observable even when the server never surfaces them as a send() failure -- the failure mode that caused generators to leak forever.

Non-disconnect messages are ignored; a well-behaved ASGI server sends only http.disconnect after the body. The sleep(0) yields control so a test double that replays the same message cannot spin the loop.

#_send_stream

async def _send_stream(send: Callable, resp: StreamResponse, request: Request | None=None) -> None

Send a streaming HTTP response, driving the generator in a child task.

The generator is iterated in a CHILD task so the request's disconnect watcher can cancel it the instant the client goes away. granian frequently never raises from send() on disconnect, so a generator blocked between yields would otherwise leak forever and its finally cleanup would never run. resp.generator.aclose() is guaranteed in every path: normal completion, generator error, send failure, and disconnect cancellation.

Send failures (client vanished mid-stream) are swallowed cleanly -- the response has already started and cannot be changed. Generator errors are NOT swallowed; they propagate to the caller (which must not attempt a second response after start).

#_accepted_params

def _accepted_params(handler: Callable) -> frozenset[str] | None

Return the set of keyword names handler accepts, or None if it takes **kwargs (accepts everything).

#_deep_convert_pydantic

def _deep_convert_pydantic(obj: Any) -> Any

Recursively convert Pydantic models to plain dicts/lists.

Walks dicts and lists, calling .model_dump(mode="json") on any object that has model_dump (i.e. Pydantic BaseModel instances).

#_send_result

async def _send_result(send: Callable, result: Any, request: Request | None=None) -> None

Dispatch a handler return value to the appropriate sender.

#_MountLifespan

Drives the ASGI lifespan protocol for one mounted sub-app.

The sub-app runs in its own task with private message queues. Per the ASGI lifespan spec, state the sub-app sets on the lifespan scope's state dict is carried into every request scope forwarded to it. Sub-apps that finish (raise or return) without sending any lifespan message are treated as not supporting lifespan, matching the server convention (e.g. uvicorn).

#startup

async def startup(self) -> str | None

Send lifespan.startup; returns an error message on failure, or None on success (including apps that don't support lifespan).

#shutdown

async def shutdown(self) -> str | None

Send lifespan.shutdown; returns an error message or None.

#_compute_static_build_id

def _compute_static_build_id(static_dir: Path | None) -> str

Return a SHA-256 build id derived from static asset file contents.

Files are visited in sorted relative-path order and their raw bytes folded into a single SHA-256 digest. The relative path is mixed in as well so that moving identical bytes between filenames changes the id. Only file contents and paths participate -- never mtime, size, or inode -- so the id is stable across touch and across machines given identical bytes.

With no static directory (or an empty one) the digest is over the empty set and equals :data:_EMPTY_BUILD_ID.

#_cache_control_for

def _cache_control_for(filename: str) -> str

Return the Cache-Control value for a served static filename.

Hashed Vite-style assets are immutable and cached for a year; everything else (assets with no content hash, index.html, SPA fallback) is no-cache so clients always revalidate.

#_serve_static

async def _serve_static(send: Callable, static_dir: Path, rel_path: str) -> bool

Serve a static file. Returns True if served, False if not found.

#_serve_spa_fallback

async def _serve_spa_fallback(send: Callable, spa_fallback: Path) -> None

Serve the SPA fallback file (typically index.html).

#AppConfig

Configuration for :func:create_app.

All fields correspond to the keyword arguments of create_app. Pass an AppConfig instance as the config parameter, and/or supply individual keyword arguments. Keyword arguments override matching fields on the config object.

#create_app

def create_app(router: Router, config: AppConfig | None=None, **kwargs: Any) -> Callable

Create an ASGI application callable.

Accepts an optional config (:class:AppConfig) and/or keyword arguments. Keyword arguments override matching fields on the config object. If neither is supplied, defaults from AppConfig are used.

If api_prefix is set (e.g. "/api"), the SPA fallback will not serve index.html for paths that start with the prefix -- they fall through to the 404 handler instead.

Built-in middleware (applied when their parameters are truthy):

• trusted_hosts: TrustedHostMiddleware (outermost)
• vite_dev_port: ViteDevProxy
• cors_origins: CORSMiddleware
• request_id: RequestIDMiddleware
• request_timing: RequestTimingMiddleware (innermost)

Custom middleware supplied via middleware wraps after built-in middleware (between the app and the built-in stack).