fastware v0.6.0 /src.fastware.app
Updated
On this page

ASGI application factory (create_app/AppConfig) with middleware chain, static-file serving, SPA fallback, lifespan hooks, DI, and WebSocket routing.

#src.fastware.app

#src.fastware.app

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

#_disconnect_watcher

python
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

python
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

python
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

python
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

python
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

python
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

python
async def shutdown(self) -> str | None

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

#_compute_static_build_id

python
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

python
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

python
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

python
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

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

Search