AsgiServer

Version: 2.0.0 Status: SOURCE OF TRUTH Last Updated: 2026-06-28

Overview

AsgiServer is the root ASGI entry point. It is a demultiplexer over mounted applications: it does no routing of its own. The server boots from a config.py, mounts one application per URL prefix, and delegates each request to the application that owns the first path segment. Each application does its own routing (via genro-routes) inside handle_request.

There is a single dispatch model — there is no “flat mode” / “router mode” switch. The server demultiplexes by first path segment; the app routes.

Inheritance

from genro_routes import RoutingClass

from .auth_mixin import BasicAuthMixin

class AsgiServer(BasicAuthMixin, RoutingClass):
    """Root ASGI entry point: a demultiplexer over the mounted apps."""

BasicAuthMixin supplies authentication; RoutingClass supplies the genro-routes routing machinery used by the mounted apps and by the server’s own plugin-config introspection helpers.

Boot Sequence (config-driven)

The server configures itself from a config.py file at construction time:

__init__ builds the fixed parts (one per instance, not configuration-driven): session store, storage, resource loader, logger, lifespan, request registry, dispatcher, WSX handler, and the built-in ServerApplication (mounted as "_server").
_load_configuration() imports config.py and instantiates its ServerConfiguration class (a subclass of AsgiConfigBuilder).
A ConfigurationHandler(self) renders that builder onto the server: the render is what mounts apps, builds the middleware chain, and sets host/port/reload.

self.config_handler = ConfigurationHandler(self)
self.config_handler.configure(self._load_configuration())

config.py must define a class named ServerConfiguration; anything else is an error. The directory containing config.py is the server’s base_dir.

Class Definition

class AsgiServer(BasicAuthMixin, RoutingClass):
    __slots__ = (
        "_auth_config", "config_path", "config_handler", "host", "port",
        "reload", "session_store", "base_dir", "storage", "resource_loader",
        "logger", "lifespan", "request_registry", "dispatcher", "wsx_handler",
        "openapi_info", "server_application", "apps", "parent", "db_registry",
    )

    def __init__(
        self,
        config_path: str | Path,
        host: str | None = None,
        port: int | None = None,
        reload: bool | None = None,
        parent: Any = None,
    ) -> None:
        ...

host, port, and reload are runtime overrides: a non-None value passed to the constructor wins over the value from the configuration.

Attributes

Attribute	Type	Description
`config_path`	`Path`	Resolved path to the `config.py` that configures this server
`base_dir`	`Path`	Directory of `config.py` (server directory)
`config_handler`	`ConfigurationHandler`	Renders the builder onto this server
`host` / `port` / `reload`	`str \| None` / `int \| None` / `bool \| None`	Runtime overrides; filled from config when left `None`
`apps`	`dict[str, AsgiApplication]`	Mounted apps keyed by mount name (the empty string `""` is the main app)
`dispatcher`	`ASGIApp`	Demultiplexes HTTP requests to the target app (wrapped by the middleware chain)
`wsx_handler`	`WsxHandler`	Handles WebSocket connections (WSX protocol)
`lifespan`	`ServerLifespan`	Manages startup/shutdown
`server_application`	`ServerApplication`	System endpoints, mounted as `"_server"`
`session_store`	`MemorySessionStore`	Server-managed session store
`storage`	`LocalStorage`	Local storage service rooted at `base_dir`
`resource_loader`	`ResourceLoader`	Hierarchical app/server resource fallback
`request_registry`	`RequestRegistry`	Tracks the current request
`db_registry`	`dict[str, Any]`	Registered database handlers by name
`openapi_info`	`dict[str, Any]`	OpenAPI metadata
`logger`	`Logger`	Server logger (`"genro_asgi"`)
`parent`	`Any`	Optional parent object with shared resources/logic

Dispatch Model

Apps are mounted under bare segment names (no leading/trailing slashes), and requests are demultiplexed by the first path segment:

/api/users/123  → mount "api"  → apps["api"]
/users/123      → mount "users" → apps["users"]  (or apps[""] if no such mount)
/               → mount ""      → apps[""]

The empty mount "" is the main app: it claims any path no other app owns.

The dispatch logic lives in Dispatcher (server/dispatcher.py):

def _resolve_mount(self, path: str) -> str:
    stripped = path.strip("/")
    if "/" in stripped:
        return stripped.split("/", 1)[0]
    return stripped

async def __call__(self, scope, receive, send) -> None:
    if "_headers" not in scope:
        scope["_headers"] = {
            name.decode("latin-1").lower(): value.decode("latin-1")
            for name, value in scope.get("headers", [])
        }
    path = scope.get("path", "/")
    mount = self._resolve_mount(path)
    app = self.server.apps.get(mount) or self.server.apps.get("")
    if app is None:
        from ..exceptions import HTTPNotFound
        raise HTTPNotFound(path)
    await app(scope, receive, send)

The chosen app is invoked as an ASGI callable: it enters its own middleware chain (if any) and ends at its own handle_request. The dispatcher only passes the baton — it does not rewrite scope["path"] or scope["root_path"].

Applying Configuration (the server is the render target)

The configuration handler renders ConfigNodes onto the server; each top-level node is dispatched by tag to a _apply_<tag> method. An unknown tag raises (grammar and server must stay in sync):

def apply_configuration(self, nodes: list[Any]) -> None:
    for node in nodes:
        method = getattr(self, f"_apply_{node.tag}", None)
        if method is None:
            raise ValueError(f"unknown configuration section '{node.tag}'")
        method(node)

Recognized sections:

Tag	Method	Effect
`server`	`_apply_server`	Set `host`/`port`/`reload` (a constructor override wins)
`middleware`	`_apply_middleware`	Build the global middleware chain wrapping a `Dispatcher(self)` and assign it to `self.dispatcher`
`authMiddleware`	`_apply_authMiddleware`	Re-init `BasicAuthMixin` from bearer/basic/jwt attributes
`applications`	`_apply_applications`	Mount every app of the collection on its derived mount
`openapi`	`_apply_openapi`	Set `openapi_info`
`databases`	`_apply_databases`	Register each `database` child in `db_registry`

Mounting applications from config

def _apply_applications(self, node: Any) -> None:
    default = node.attrs.get("default")
    for child in node.children:
        code = child.label
        mount = child.attrs.get("mount")
        if mount is None:
            mount = "" if code == default else code
        self._mount_app(child, mount)

Each child’s code (its collection key / label) determines the mount: the explicit mount attribute if given, else the empty mount "" when the code equals the collection’s default, else the code itself.

def _mount_app(self, node: Any, mount: str) -> None:
    if mount in self.apps:
        raise ValueError(f"mount {mount!r} is already taken")
    attrs = dict(node.attrs)
    attrs.pop("code", None)
    attrs.pop("mount", None)
    attrs.pop("middleware", None)
    app_class = attrs.pop("app_class")
    instance = app_class(**attrs)
    instance.mount_name = mount
    self.attach_instance(instance)        # dual relationship: app.server -> self
    self.apps[mount] = instance
    instance.configure(node)              # the app interprets its own config

app_class (imported by config.py) is the application class; the remaining attributes are its constructor kwargs. The app then interprets its own configuration node in configure() — the server does not interpret an app’s config.

Mount Method (programmatic)

For programmatic mounting (outside the config flow):

def mount(self, name: str, app: Any) -> None:
    """Mount an application on the server.

    Args:
        name: Mount name (becomes the URL prefix). Must be unique.
        app: Application instance (AsgiApplication or RoutingClass).

    Raises:
        ValueError: If the name is already mounted.
    """
    if name in self.apps:
        raise ValueError(f"mount {name!r} is already taken")
    app.mount_name = name
    self.attach_instance(app)             # dual relationship: app.server -> self
    self.apps[name] = app

mount takes a bare name (not a slashed path). There is no slash normalization, no per-app RequestRegistry, and no ServerBinder — the server holds a single request_registry, and apps reach the server through the genro-routes dual relationship (app.server) established by attach_instance.

ASGI Interface

__call__ branches on the ASGI scope type:

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    if scope["type"] == "lifespan":
        await self.lifespan(scope, receive, send)
    elif scope["type"] == "websocket":
        await self.wsx_handler(scope, receive, send)
    else:                                  # http
        await self.dispatcher(scope, receive, send)

lifespan → ServerLifespan (startup/shutdown).
websocket → WsxHandler (WSX protocol over WebSocket).
http (and anything else) → self.dispatcher, which is either a bare Dispatcher(self) or the middleware chain wrapping one, depending on the middleware config section.

Databases

The databases section registers one handler per database child, keyed by its code:

def _apply_databases(self, node: Any) -> None:
    for child in node.children:
        descriptor = dict(child.attrs)
        code = descriptor.pop("code")
        self.register_db(code, self.build_db_handler(code, descriptor))

build_db_handler requires a db_class (imported by config.py) that builds the db from the remaining attributes; an optional db_handler_class (default AsgiDbHandlerBase) wraps it. Registered handlers are reachable via get_db(name) / the db property (shortcut for get_db("default")).

Plugin Config Persistence

The server exposes runtime plugin-configuration helpers built on the genro-routes public API (self.routing.configure(...)):

Method	Purpose
`get_plugin_config(router_path="", entry="")`	Introspect live plugin config (`routing.configure("?")`)
`set_plugin_config(router_path, plugin_name, target, **config_values)`	Apply a plugin config change at runtime (not persisted)
`save_plugin_config()`	Persist the current tree to `plugin_config.json` in `base_dir`
`revert_plugin_config()`	Revert runtime config to the saved file
`apply_persisted_config()`	Apply `plugin_config.json` at boot (after plugins are attached)

The persistence file is base_dir / "plugin_config.json" (plugin_config_file).

Run Method

run() takes no arguments; it reads self.host / self.port / self.reload (already filled from constructor overrides or the config):

def run(self) -> None:
    """Run the server using Uvicorn."""
    import os
    import uvicorn

    host = self.host or "127.0.0.1"
    port = self.port or 8000
    reload = self.reload or False

    self.logger.info(f"Starting server on {host}:{port}")
    if reload:
        os.environ["GENRO_ASGI_CONFIG"] = str(self.config_path)
        uvicorn.run(
            "genro_asgi.server.server:server_factory",
            host=host, port=port, reload=True,
            reload_dirs=[str(self.base_dir)], factory=True,
        )
    else:
        uvicorn.run(self, host=host, port=port)

In reload mode uvicorn needs an import string and a fresh instance per restart, so the config path is passed via the GENRO_ASGI_CONFIG environment variable to the module-level server_factory(), which rebuilds an AsgiServer from it.

Usage

from genro_asgi import AsgiServer

# Boot from a config.py (its directory is the server dir).
server = AsgiServer("config.py")
server.run()

# Runtime overrides win over the configuration:
server = AsgiServer("config.py", host="0.0.0.0", port=9000, reload=True)
server.run()

Error Handling

When no app claims the path (and there is no empty-mount fallback app), the dispatcher raises HTTPNotFound (a 404 HTTPException subclass). Router-level errors raised inside an app are mapped to HTTP exceptions by the app’s own dispatch layer (see 00-overview.md and the per-app routing docs): not_found → 404, not_authorized → 403, not_authenticated → 401, not_available → 503, validation_error → 400.

Architecture Diagram

┌─────────────┐         ┌──────────────────────────────────────────────────────────┐
│   Uvicorn   │         │  AsgiServer (BasicAuthMixin, RoutingClass)               │
│   :8000     │ ──────► │  __call__ branches on scope["type"]:                     │
│             │         │    lifespan  → ServerLifespan                            │
│             │         │    websocket → WsxHandler (WSX)                          │
│             │         │    http      → dispatcher (middleware chain → Dispatcher)│
│             │         │                                                          │
│             │         │  Dispatcher demultiplexes by first path segment:         │
│             │         │    /api/...   → apps["api"]                              │
│             │         │    /...       → apps[""]   (empty-mount = main app)      │
│             │         │    /_server/… → apps["_server"] (ServerApplication)      │
│             │         │  each app does its own routing in handle_request         │
└─────────────┘         └──────────────────────────────────────────────────────────┘