Core

The server, the application contract, request/response, and the shared types.

Server

The base server: primary app, secondary mounts, ASGI dispatch, uvicorn boot.

BaseServer is the common substrate of every server (SPECIFICATION.md §4, D2): it REQUIRES the primary application (primary= kwarg; missing raises TypeError) and keeps secondary applications in a mounts dict keyed by mount_name — the one demux mechanism of D3. At the base, authenticate() answers nobody (None) and session() answers none (None). It owns exactly one thread pool (D2): run_sync() dispatches a blocking handler onto it via loop.run_in_executor while async handlers stay on the loop; the pool is provisioned lazily on first use and torn down at shutdown.

As an ASGI callable, __call__ dispatches on the scope type: http runs the D3 demux (first path segment → matching mount with that segment stripped, otherwise the primary with the full path); websocket runs on_websocket, whose DEFAULT is the empty socket of D7 (accepts nothing, closes cleanly with code 1000); lifespan runs the Lifespan handler (ordered startup, reverse shutdown, error isolation). Each http dispatch is registered in the RequestRegistry (requests) for the span of the request — the current request and the in-flight picture. serve() boots uvicorn programmatically.

Cooperative init (D16): peels its own kwargs (primary, max_threads) and, as the end of the chain, raises TypeError naming any leftover kwargs. Mixins go BEFORE BaseServer in the MRO.

Ownership channel (one direction): attaching the primary and mount() both assign app.server = self; the app-side setter enforces exactly-once.

class genro_asgi.server.BaseServer(**kwargs)[source]

Bases: object

Base server owning the primary application and the secondary mounts.

Constructor kwargs peeled here: primary — the always-present primary application (D2), answering / and everything no mount claims — and max_threads — the pool’s worker count, handed to WorkPool (None keeps the stdlib default).

Parameters:

kwargs (Any)

property primary: BaseApplication

answers / and the unclaimed rest.

Type:

The primary application

property mounts: dict[str, BaseApplication]

Secondary applications keyed by mount_name (may be empty).

mount(app)[source]

Register app as a secondary mount keyed by its mount_name.

Assigns the ownership channel (app.server = self). A secondary without a mount_name cannot be demuxed and a claimed name cannot be claimed twice: both raise ValueError.

Return type:

None

Parameters:

app (BaseApplication)

property databases: dict[str, Any]

Database handlers keyed by their config code (may be empty).

add_database(code, handler)[source]

Register handler under code. A claimed code raises ValueError.

Return type:

None

Parameters:
property lifespan: Lifespan

The Lifespan handler managing this server’s startup/shutdown.

property pool: WorkPool

The server’s single thread pool for blocking (sync) handlers.

property requests: RequestRegistry

The registry of in-flight requests and the current one.

async run_sync(fn, *args)[source]

Dispatch blocking fn onto the pool (the app-side sync protocol).

Apps call self.server.run_sync(...) for blocking work so it runs off the event loop; async handlers simply stay on the loop and never touch the pool.

Return type:

Any

Parameters:
authenticate(request)[source]

Base answer: nobody (None). Auth capabilities override this.

Return type:

Any

Parameters:

request (Any)

session(request)[source]

Base answer: none (None). Session capabilities override this.

Return type:

Any

Parameters:

request (Any)

demux(scope)[source]

D3 demux: pick the app for an http scope and the scope it receives.

First path segment matching a mount → that app, with the segment stripped from path (the forwarded path is rebuilt from the same remainder used to find the segment, so //api/x forwards /x); anything else (including /) → the primary with the full path unchanged.

Return type:

tuple[BaseApplication, MutableMapping[str, Any]]

Parameters:

scope (MutableMapping[str, Any])

async on_websocket(scope, receive, send)[source]

The empty websocket socket (D7): consume the connect, close cleanly.

The base accepts nothing and closes with code 1000. The websocket motor (Q1) overrides this hook in a later macro.

Return type:

None

Parameters:
property uvicorn_server: Server | None

The uvicorn Server once serve() has built it (else None).

Callers that boot the server in a background thread read the bound port from uvicorn_server.servers[0].sockets[0].getsockname() after uvicorn_server.started turns true.

serve(host='127.0.0.1', port=0)[source]

Boot uvicorn programmatically, serving this server (blocking).

Builds uvicorn.Config/uvicorn.Server and runs it. port=0 lets the OS assign an ephemeral port, discoverable via uvicorn_server once started.

Return type:

None

Parameters:

AsgiServer — the shipped mono-process server composition (D22, D6, D16).

AsgiServer stacks every core capability mixin over BaseServer in one MRO (CommunicationMixin, AuthMixin, SessionMixin, MiddlewareMixin, PluginMixin, StorageMixin, TaskMixin, BaseServer): the complete mono-process async server of D22. TaskMixin sits after StorageMixin (it needs server.storage) and before BaseServer (its lifespan hook must wrap the base Lifespan). The future internal (worker) server simply composes the SAME base WITHOUT the auth mixin (D6 by construction — the base never learned about the chain).

Its cooperative __init__ peels the kwargs the frozen Macro 1 BaseServer does not accept — host/port/external_url plus server_app (the _server config-class lift) — and forwards everything else (primary, auth, session_store/session_ttl, middleware/middleware_registry, plugins/plugin_registry, storage/storage_key, parent) down the D16 chain. The peeled host/port become the defaults of serve, so a config-built server serves on its configured address unless the caller overrides it; server_app travels to _mount_server_app.

host/port are the LISTENER; external_url is the server’s PUBLIC base address — the two differ behind a proxy and answer different questions. The listener says where to bind; the public address is what the server calls itself when it hands its own URL to a third party. Only one consumer needs it today (an OIDC provider is given an absolute redirect_uri, RFC 6749 §3.1.2), and it is DECLARED rather than derived from a request: the URI must match the one registered with the provider — a deployment fact known to whoever installs — and deriving it from the client-supplied Host would build a value the provider then rejects. Missing it with a provider configured is a boot error (_check_oidc_external_url), not an opaque provider error at the first login.

Once the chain has run, __init__ mounts the automatic _server app (_mount_server_app, D4 “automatic, not configured”): a hand-built AsgiServer(primary=...) exposes /_server/... exactly like a config-materialized one, and ConfigurationHandler.materialize never special-cases it.

class genro_asgi.asgi_server.AsgiServer(**kwargs)[source]

Bases: CommunicationMixin, AuthMixin, SessionMixin, MiddlewareMixin, PluginMixin, StorageMixin, TaskMixin, BaseServer

The shipped composition: communication + auth + sessions + chain + plugins + storage + base.

Constructor kwargs peeled here: host and port — the serve defaults carried from the config’s server section — external_url, the server’s public base address (trailing slash stripped), plus server_app, the _server config-class lift forwarded to the auto-mounted app. Every other kwarg flows to the capability mixins and the base (D16 cooperative init).

Parameters:

kwargs (Any)

property login_enabled: bool

True when the _server app carries a registered auth method.

The challenge negotiation (ErrorMiddleware) reads this to decide whether a 401 becomes a login redirect (browser) or a login_url body (API). It reflects live state: ServerApplication registers the password method at construction, so its server has a login surface.

property config_host: str | None

The host from the config’s server section (None if unset).

property config_port: int | None

The port from the config’s server section (None if unset).

property external_url: str | None

The server’s public base URL, without a trailing slash (None if unset).

What the server calls ITSELF when it hands its own address to a third party — distinct from the host/port it binds to, which differ behind a proxy. Declared in the config’s server section; the only consumer today is the OIDC redirect_uri, which must be absolute.

serve(host=None, port=None)[source]

Boot uvicorn, defaulting host/port to the configured values.

The caller’s explicit host/port win; otherwise the config’s server section is used, falling back to the BaseServer defaults (127.0.0.1 and an OS-assigned port).

Return type:

None

Parameters:

  • host (str | None)

  • port (int | None)

Applications (contract)

App-side contract: the base class every mountable application extends.

BaseApplication is what the server requires of an app (SPECIFICATION.md §4, D7): an ASGI callable (__call__ implemented by concrete subclasses) with a mount_name (constructor kwarg, empty for a primary), a server property assigned exactly once by the owning server at attach time (ownership channel, one direction — a second assignment raises RuntimeError), and lifecycle hooks on_startup/on_shutdown that subclasses may override as sync OR async (the caller detects which at call time).

Cooperative init (D16): every class in the family implements __init__(self, **kwargs), peels ITS OWN kwargs and forwards the rest via super().__init__(**rest). Mixins go BEFORE the base in the MRO; this base is the end of the chain and raises TypeError naming any leftover kwargs.

class genro_asgi.application.BaseApplication(**kwargs)[source]

Bases: object

Base class for applications attached to a BaseServer.

Constructor kwargs peeled here: mount_name — the URL prefix under which a server mounts this app as a secondary (empty for a primary).

Parameters:

kwargs (Any)

property mount_name: str

URL prefix under which this app is mounted (empty for a primary).

property server: BaseServer | None

The server that owns this app (None until attached).

on_startup()[source]

Lifecycle hook run at server startup. Override as sync or async.

Return type:

None

on_shutdown()[source]

Lifecycle hook run at server shutdown. Override as sync or async.

Return type:

None

RoutedApplication: the application base wiring genro-routes into the core.

RoutedApplication composes the app-side contract (BaseApplication) with the genro-routes RoutingClass: handlers are @route-decorated methods on subclasses, and external RoutingClass instances mount as sub-trees via add_branches({"name": ..., "instance": child}). The constructor peels db_name (the request.db seam resolves it against the server’s database registry) and plugs the auth plugin on the app router, so entries declaring auth_rule are filtered by the request’s authorization tags — attached children inherit the plug.

The config-driven plugins (the openapi dialect today) are armed LAZILY: on the first route access made after the app is attached to a server, the app calls server.arm_router(self.route) (once, guarded), so a server built with a plugins section plugs them onto every routed app it hosts. A composition whose server lacks the PluginMixin exposes no arm_router and arms nothing — the app degrades to the auth plug alone.

The ASGI dispatch (__call__) is the per-app routing engine: build a Request bound to this app, eager-parse it (init), resolve the node from the request path — already mount-relative, the server demux strips the prefix (D3) — with the identity tags of scope["auth"] as auth filters, bind kwargs, execute (async handlers stay on the loop, sync handlers go through server.run_sync — the Macro 1 pool protocol), then answer via request.response.set_result(value, metadata). Resolution failures raise core exceptions (ROUTER_ERRORS: unknown or unavailable path → HTTPNotFound; a ruled entry denied, for missing or mismatched tags → HTTPForbidden) that propagate to the server’s ErrorMiddleware. Invalid handler arguments become HTTPBadRequest (400) — never a 500: genro-routes channels every bad-argument error (a pydantic.ValidationError and an unbindable-argument TypeError alike) through the node’s single validation_error exception mapping, so the dispatcher catches one marker. A TypeError raised inside a SYNC handler body is indistinguishable from a binding error and maps to 400 too; an async body’s TypeError surfaces at await time and reaches ErrorMiddleware as a 500. There is no local cleanup drain: the server finally owns end-of-request cleanups.

Kwargs binding: bind_kwargs starts from request.handler_kwargs() (query + body by content-type) and reconciles a hydrated JSON body with a scalar-parameter handler — when the node’s neutral params block declares fields (pydantic plugin) and the handler does not itself absorb body_data or **kwargs, the body dict is spread over the declared names through spread_over_params (extras dropped). Auth without the middleware: when scope carries no auth key the resolution runs unfiltered — the public router exposes exactly what the auth plugin leaves untagged.

class genro_asgi.routed_application.RoutedApplication(**kwargs)[source]

Bases: BaseApplication, RoutingClass

Application base serving @route handlers through the app router.

Constructor kwargs peeled here: db_name — the server database code the request.db seam resolves for this app (None falls back to "default"). The rest flows down the D16 chain (mount_name to BaseApplication).

Parameters:

kwargs (Any)

property db_name: str | None

Server database code for the request.db seam (None → default).

property route: Router

The app router; arms the server’s configured plugins on first access.

The first access made once a server owns this app triggers server.arm_router (once, guarded by _armed); accesses before attachment — the auth plug in __init__ — arm nothing.

auth_filters(scope)[source]

Auth filters for node resolution, from the scope identity.

An Avatar on scope["auth"] becomes the comma-separated auth_tags the auth plugin evaluates entry rules against. No identity — key absent (middleware off) or None (anonymous) — passes no filter: the plugin still denies every ruled entry.

Return type:

dict[str, str]

Parameters:

scope (MutableMapping[str, Any])

make_callable(node, request)[source]

Package the node invocation as the zero-arg call the dispatcher runs.

The node declares its handler’s nature (genro-routes marks async entries so asyncio.iscoroutinefunction(node) is honest); the dispatcher reads that same nature to pick the vehicle — the returned async callable is awaited on the loop, the sync one goes through the server pool. bind_kwargs decides the arguments.

Return type:

Callable[[], Any]

Parameters:

  • node (RouterNode)

  • request (Request)

bind_kwargs(node, request)[source]

Reconcile the request kwargs with the handler’s declared parameters.

Base: request.handler_kwargs(). A hydrated JSON body arrives as a single body_data dict; a REST handler declares scalar parameters, so the dict is spread over the fields the handler accepts (from the node’s neutral params block — never inspect). The whole body_data is kept when the handler itself declares it, accepts **kwargs, or exposes no signature (no pydantic plugin).

Return type:

dict[str, Any]

Parameters:

  • node (RouterNode)

  • request (Request)

spread_over_params(node, data)[source]

Fit a dict of values to the handler’s declared parameters.

Shared by every wire dialect (REST body today, MCP arguments later): keeps data whole when the handler declares no signature (fields is None — no pydantic plugin) or accepts **kwargs; otherwise keeps only the declared names, dropping extras. An empty fields list is a known no-parameter handler: everything drops.

Return type:

dict[str, Any]

Parameters:

Request and response

HTTP request: one flat class over the ASGI scope, eager body parsing.

Request is HTTP-only — no transport abstraction (the WSX/message transport is orchestration, out of the core). It wraps the ASGI scope and, in the async init(), parses headers/cookies/query/body once via genro_tytx.asgi_data (which hydrates JSON/XML/msgpack bodies and hands back the raw bytes for anything else). TYTX mode is detected from the X-TYTX-Transport header; the paired Response reads tytx_mode / tytx_transport to serialize the reply in the same transport.

The owning application creates it (Request(scope, receive, application=app), or server= directly) and holds the response seam: self.response is a Response bound back to this request (the TYTX path).

handler_kwargs() builds the kwargs a route handler receives: the query is the base; a form body (x-www-form-urlencoded) is decoded and merged (body wins on a clash), a hydrated body is passed whole as body_data, opaque bytes as body_raw, an empty body adds nothing.

db is the deferred preparation layer (no ORM yet): on first access it resolves the server’s registered handler for the owning app’s db_name (else "default") and registers its closeConnection as a request cleanup (drained by the server at end of request). get_db(name) is a plain lookup with no cleanup registration. Auth and session ride the scope (scope["auth"] — an Avatar or None — and scope["session"]), set by the middleware chain.

class genro_asgi.request.Request(scope, receive, *, server=None, application=None)[source]

Bases: object

An ASGI HTTP request: scope wrapper with eager, TYTX-aware body parsing.

Parameters:
async init()[source]

Parse headers, cookies, query and body from the scope (once).

Delegates to genro_tytx.asgi_data and then derives TYTX mode, the request id (x-request-id header or a fresh uuid4) and the optional client correlation id (x-external-id).

Return type:

None

property id: str

the x-request-id header, or a generated uuid4.

Type:

Correlation id

property method: str

HTTP method (uppercased).

property path: str

Request path.

property headers: dict[str, Any]

Request headers (lowercase keys), values hydrated by TYTX.

property cookies: dict[str, str]

Request cookies parsed from the Cookie header.

property query: dict[str, Any]

Query parameters (typed via TYTX).

property data: Any

hydrated value, raw bytes, or None when empty.

Type:

Parsed body

property content_type: str | None

Content-Type header value, or None.

property external_id: str | None

Client-provided correlation id (x-external-id header).

property tytx_mode: bool

True when the request declared a TYTX transport.

property tytx_transport: str | None

TYTX transport (json/xml/msgpack), or None.

property created_at: float

Wall-clock timestamp captured at construction.

property age: float

Seconds elapsed since construction.

property scope: MutableMapping[str, Any]

The raw ASGI scope.

property server: BaseServer | None

The owning server (passed directly, or via the owning application).

property application: BaseApplication | None

The application that created this request (None if unbound).

property avatar: Any

The identity acting on this request (an Avatar) or None.

The effective identity the auth chain resolved for this request — header credentials or the session’s avatar — read from the scope.

property auth_tags: list[str]

Authorization tags of the current identity (empty when anonymous).

property session: Any

The session object attached by SessionMiddleware, or None.

property db: Any

The default db handler for the owning app, or None (lazy).

Resolves server.databases[name] where name is the owning application’s db_name attribute if set, else "default". On the first successful resolution it registers handler.closeConnection as a request cleanup (drained by the server at end of request). Returns None when there is no server or no handler under that name.

Preparation layer only: no pooling, no transactions, no per-app registry.

get_db(name)[source]

Look up a registered db handler by name (no cleanup registration).

Return type:

Any

Parameters:

name (str)

handler_kwargs()[source]

Build the kwargs a route handler is called with (query + body).

The query params are the base. The body adds to them by content-type, not by Python shape: an x-www-form-urlencoded body arrives already hydrated (asgi_data decodes it, typed, via TYTX from_qs) and is merged — the body wins on a name clash; a hydrated body (JSON/XML/msgpack) is passed whole as body_data; opaque bytes are passed as body_raw; an empty body adds nothing.

Return type:

dict[str, Any]

HTTP response: one flat, buffered, TYTX-aware class.

Response is a single slotted class — no subclass hierarchy (no JSON/HTML/Streaming/File variants). It buffers the body in memory and, as an ASGI application, emits exactly two messages (http.response.start + http.response.body). It can be built with content or created empty and configured through set_header/set_cookie/set_result/set_error before being sent.

set_result dispatches by result type: dict/list → JSON bytes via genro_tytx.json_dumps (or TYTX serialization — media type from genro_tytx.TRANSPORT_MIME — when the bound request is in TYTX mode), Path → file bytes, bytes → as-is, str → UTF-8 text, None → empty. set_error maps an exception to a status: HTTPException subtypes carry their own status; ValueError/TypeError → 400, FileNotFoundError → 404, PermissionError → 403, anything else → 500 (logged). set_cookie appends a set-cookie header.

The request binding is optional (request=None); every request-dependent branch (the TYTX path) guards for its absence.

class genro_asgi.response.Response(content=None, status_code=200, headers=None, media_type=None, request=None)[source]

Bases: object

Buffered HTTP response, usable directly as an ASGI application.

Example

>>> response = Response(content="Hello", media_type="text/plain")
>>> await response(scope, receive, send)

# Or create empty and configure: >>> response = Response() >>> response.set_header(“X-Custom”, “value”) >>> response.set_result({“data”: 123}) # auto-detects JSON >>> await response(scope, receive, send)

Parameters:

  • content (bytes | str | None)

  • status_code (int)

  • headers (HeadersInput)

  • media_type (str | None)

  • request (Any)

__init__(content=None, status_code=200, headers=None, media_type=None, request=None)[source]

Build a response.

Note

Status 204 (No Content) and 304 (Not Modified) must not carry a body per RFC 7230; providing content with those codes may be rejected or truncated by the ASGI server.

Parameters:
Return type:

None

set_header(name, value)[source]

Append a response header. Usable before set_result.

Return type:

None

Parameters:

Append a set-cookie header (the value is URL-encoded).

Return type:

None

Parameters:
set_result(result, metadata=None)[source]

Set the body from a handler result, dispatching by type.

dict/list → JSON bytes (genro_tytx.json_dumps), or TYTX bytes/text when the bound request is in TYTX mode; Path → file bytes; bytes → as-is; str → UTF-8 text; None → empty; anything else → its str. A media_type in metadata overrides the type-based default.

Return type:

None

Parameters:
set_error(error)[source]

Set the response as an error, mapping the exception to a status.

HTTPException subtypes carry their own status; other types are looked up in ERROR_MAP (default 500, logged). The body is the {"error": <message>} document through set_result.

Return type:

None

Parameters:

error (Exception)

Streaming HTTP response: a chunked ASGI sibling of Response.

Response (response.py) is flat and BUFFERED — two ASGI messages, the whole body in memory. A stream is a different shape, not a variant, so it is a separate slotted class rather than a subclass: StreamingResponse sends the http.response.start once, then one http.response.body per chunk with more_body=True, and a terminal empty body with more_body=False. It has NO set_result (the buffered type-dispatch is deliberately not carried over): the body is an async iterator of bytes the caller supplies.

The iterator is the whole contract — a plain async for over user chunks, an SseStream (sse.py), or any bounded event source. This class only frames the ASGI message sequence around it; backpressure and heartbeats live in the source.

class genro_asgi.streaming.StreamingResponse(body_iterator, status_code=200, headers=None, media_type=None)[source]

Bases: object

Chunked HTTP response, usable directly as an ASGI application.

Example

>>> async def chunks():
...     yield b"one"
...     yield b"two"
>>> response = StreamingResponse(chunks(), media_type="text/plain")
>>> await response(scope, receive, send)
Parameters:

  • body_iterator (AsyncIterable[bytes])

  • status_code (int)

  • headers (HeadersInput)

  • media_type (str | None)

__init__(body_iterator, status_code=200, headers=None, media_type=None)[source]

Build a streaming response over an async iterator of byte chunks.

Parameters:
Return type:

None

set_header(name, value)[source]

Append a response header (before the response is sent).

Return type:

None

Parameters:

Server-Sent Events framing over StreamingResponse.

An event is a small dict — {"data": ..., "event": ..., "id": ...} — framed into the text/event-stream wire format: id:/event:/data: lines (data split across lines on newlines, per the spec), retry: once at the start when configured, and a : keepalive comment when the source falls silent longer than the heartbeat interval (the comment keeps proxies from closing an idle connection; the client ignores it). Each event ends with a blank line. data that is not a string is JSON-encoded.

The framing is shaped like channel/frame.py (a slotted codec, its own wire format) but has no bytes in common — SSE is a text protocol over HTTP, not the length-prefixed wsx envelope. SseStream is SELF-CONTAINED: it wraps ANY async source of event dicts (a user generator, a task hub subscription) and yields wire bytes; the source is the caller’s concern. Resumability (Last-Event-ID → a snapshot baseline then the live source) is built by the consumer that owns the event source, not here.

class genro_asgi.sse.SseStream(source, *, retry_ms=None, keepalive_seconds=15.0)[source]

Bases: object

Frames an async source of event dicts into text/event-stream bytes.

Note

The stream is bound to one source (dual relationship: self.source). Iterating it yields wire bytes; response() wraps it in a StreamingResponse with the SSE headers already set.

Parameters:

  • source (AsyncIterable[dict[str, Any]])

  • retry_ms (int | None)

  • keepalive_seconds (float)

__init__(source, *, retry_ms=None, keepalive_seconds=15.0)[source]

Bind to an async source of event dicts.

Parameters:

  • source (AsyncIterable[dict[str, Any]]) – Any async iterable of events; each event is a dict with an optional id/event and a data payload.

  • retry_ms (int | None) – When set, a retry: line is emitted once at the start (the client’s reconnection delay).

  • keepalive_seconds (float) – Idle interval after which a : keepalive comment is sent to hold the connection open.

Return type:

None

frame(event)[source]

Encode one event dict into an SSE record (ends with a blank line).

Return type:

bytes

Parameters:

event (dict[str, Any])

response(status_code=200)[source]

A StreamingResponse over this stream with the SSE headers set.

Return type:

StreamingResponse

Parameters:

status_code (int)

Registry, lifespan and pool

Request registry: the current request and the in-flight picture.

RequestRegistry is held by the server as a dual parent-child (self.server, SPECIFICATION.md §4) and is the SINGLE writer of the in-flight set (D12 spirit): the server registers a request on entry and unregisters it on exit, around the http dispatch. Each registration is a lightweight RegisteredRequest record (D18: slotted, high cardinality) carrying a monotonic id, the scope type, the path, the start time, and the request’s cleanup callbacks. The server drains those cleanups in the http finally (run_cleanups) so any app — routed or bare — gets end-of-request teardown (e.g. request.db closing its connection) for free.

The “current request” is exposed through a ContextVar that lives on the registry INSTANCE (never at module level): register sets it and keeps the reset token, unregister resets it. Because the ContextVar is an instance attribute, deleting the server garbage-collects everything (instance-isolation rule) and concurrent requests — each on its own task context — see their own current.

class genro_asgi.registry.RegisteredRequest(request_id, scope_type, path)[source]

Bases: object

One in-flight request tracked by the registry (D18: slotted record).

A snapshot taken at registration: the monotonic request_id, the ASGI scope_type, the path, and started_at (time.monotonic()). Slotted because requests are high cardinality.

It also owns the request’s end-of-life cleanups: add_cleanup(fn) queues a zero-arg callback and run_cleanups() drains them LIFO at the end of the dispatch (the server calls it in the http finally). The _cleanups list is lazy — allocated only when the first callback is queued — so a request that registers none pays nothing.

Parameters:

  • request_id (int)

  • scope_type (str)

  • path (str)

property request_id: int

Monotonic id assigned by the registry at registration.

property scope_type: str

ASGI scope type of the request (http).

property path: str

Request path at registration time.

property started_at: float

time.monotonic() captured at registration.

add_cleanup(callback)[source]

Queue a zero-arg callback to run at end of request (LIFO).

Return type:

None

Parameters:

callback (Callable[[], Any])

run_cleanups(error=None)[source]

Run queued cleanups LIFO, isolating and logging each one’s exception.

Called by the server in the http finally — so cleanups run whether the request succeeded or failed. error carries the terminating exception (None on success) for error-aware cleanups; the base drain runs every callback regardless.

Return type:

None

Parameters:

error (BaseException | None)

class genro_asgi.registry.RequestRegistry(server)[source]

Bases: object

Tracks in-flight requests and the current one, owned by the server.

The server is the single writer: it calls register(scope) on request entry and unregister(item) on exit. current reads the instance-owned ContextVar; in_flight counts the live requests; snapshot() lists them.

Parameters:

server (BaseServer)

property current: RegisteredRequest | None

The request being handled in this task’s context, or None.

property in_flight: int

How many requests are registered right now.

register(scope)[source]

Register a request from scope, set current, return the item.

Return type:

RegisteredRequest

Parameters:

scope (MutableMapping[str, Any])

unregister(item)[source]

Drop item from the in-flight set and reset current.

Return type:

None

Parameters:

item (RegisteredRequest)

snapshot()[source]

A list of the currently in-flight requests (registration order).

Return type:

list[RegisteredRequest]

ASGI lifespan protocol: ordered startup, reverse shutdown, error isolation.

Lifespan is constructed with the server it manages (dual parent-child: self.server, SPECIFICATION.md §4). On lifespan.startup it runs on_startup on the primary first, then on the mounted apps in mount order; on lifespan.shutdown it runs on_shutdown in REVERSE order. Hooks may be sync or async, detected with inspect.iscoroutinefunction at call time.

A hook that raises is logged and the sequence CONTINUES: one app’s error never blocks the others, and uvicorn always receives the matching .complete message — app errors are isolated, never abort the protocol.

class genro_asgi.lifespan.Lifespan(server)[source]

Bases: object

ASGI lifespan handler, held by the server as a dual parent-child.

Parameters:

server (BaseServer)

async startup()[source]

Run on_startup: primary first, then mounts in mount order.

Return type:

None

async shutdown()[source]

Run on_shutdown in REVERSE order.

Return type:

None

The server’s single thread pool for blocking work (SPECIFICATION.md §4, D2).

WorkPool wraps one concurrent.futures.ThreadPoolExecutor and is held by the server as a dual parent-child (self.server). Async handlers stay on the event loop; only sync handlers reach the pool, dispatched through BaseServer.run_sync via loop.run_in_executor.

Lazily provisioned (invariant #1 — build lazily on the running loop): the executor is created on the first dispatch, never at boot, and torn down at lifespan shutdown only if it was ever provisioned.

class genro_asgi.pool.WorkPool(server, max_threads=None)[source]

Bases: object

One thread pool for blocking (sync) handlers, owned by the server.

Constructor kwarg: max_threads — the executor’s max_workers (None uses the stdlib default, min(32, cpu + 4)). Threads are named genro-pool* so a handler can assert it ran off the loop.

Parameters:
property provisioned: bool

Whether the executor exists yet (a sync dispatch has happened).

property executor: ThreadPoolExecutor

The pool’s executor, created on first access (lazy provisioning).

async run(fn, *args)[source]

Run blocking fn on a pool thread, provisioning on first call.

The caller’s context is copied into the worker thread (what asyncio.to_thread does), so a sync handler sees the loop-side ContextVars — e.g. the registry’s current request.

Return type:

Any

Parameters:
shutdown(wait=True)[source]

Tear the executor down — a no-op if it was never provisioned.

Resets the lazy slot so a later dispatch re-provisions: a server reused through repeated serve() rounds keeps working.

Return type:

None

Parameters:

wait (bool)

Exceptions and types

HTTP control-flow exceptions: raised anywhere, answered by the errors middleware.

Plain classes, no framework machinery: HTTPException(status, detail=None, headers=None) carries the response status (an optional plain-text detail and optional response headers — ASGI (name, value) byte pairs forwarded to the response, e.g. a WWW-Authenticate challenge on a 401); the common errors are pre-filled subclasses — HTTPBadRequest (400), HTTPNotFound (404), HTTPUnauthorized (401), HTTPForbidden (403). Redirect(location, status=302) is the redirecting sibling: its location becomes the Location header. The mapping to actual ASGI responses lives in middleware/errors.py.

exception genro_asgi.exceptions.HTTPBadRequest(detail=None, headers=None)[source]

Bases: HTTPException

400 Bad Request.

Parameters:
Return type:

None

exception genro_asgi.exceptions.HTTPException(status, detail=None, headers=None)[source]

Bases: Exception

HTTP error carried as an exception: status, optional detail and headers.

headers are ASGI (name, value) byte pairs the errors middleware forwards onto the response (e.g. a WWW-Authenticate challenge).

Parameters:
Return type:

None

exception genro_asgi.exceptions.HTTPForbidden(detail=None, headers=None)[source]

Bases: HTTPException

403 Forbidden.

Parameters:
Return type:

None

exception genro_asgi.exceptions.HTTPNotFound(detail=None, headers=None)[source]

Bases: HTTPException

404 Not Found.

Parameters:
Return type:

None

exception genro_asgi.exceptions.HTTPUnauthorized(detail=None, headers=None)[source]

Bases: HTTPException

401 Unauthorized.

Parameters:
Return type:

None

exception genro_asgi.exceptions.Redirect(location, status=302, headers=None)[source]

Bases: HTTPException

HTTP redirect: location becomes the Location header.

Parameters:
Return type:

None

ASGI type aliases for genro-asgi.

Aliases follow the ASGI spec (MutableMapping, not TypedDict, for extensibility): Scope — connection metadata (type, method, path, headers, …); Message — messages between app and server (the type key identifies the message); Receive/Send — the two async channel callables; ASGIApp — the standard application signature.