# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""ASGI Server - main entry point for genro-asgi applications.
Classes:
AsgiServer -- central coordinator: reads a ``config.py`` (a
ServerConfiguration builder), is the render target that
configures itself, mounts apps, builds the middleware
chain, handles lifespan, demultiplexes requests to the
mounted apps via the Dispatcher (each app routes itself).
Inherits BasicAuthMixin and RoutingClass. Run with
``AsgiServer(config_path).run()``.
WebSocket connections are handled by WsxHandler (WSX protocol over WebSocket).
"""
from __future__ import annotations
import importlib.util
import json
import logging
import os
from pathlib import Path
from typing import Any
import uvicorn
from .auth_mixin import BasicAuthMixin
from .dispatcher import Dispatcher
from ..applications import AsgiApplication
from ..authentication.auth_method import PasswordMethod
from ..authentication.oidc_method import OidcMethod
from ..authentication.user_store import FileUserStore
from ..db import AsgiDbHandlerBase
from ..lifespan import ServerLifespan
from ..session import MemorySessionStore, Session
from ..middleware import middleware_chain
from ..resources import ResourceLoader
from ..response import Response
from ..request import RequestRegistry, BaseRequest
from .server_applications.server_application import ServerApplication
from .server_applications.plugin_config import PluginConfigSection
from .server_applications.users import UsersSection
from ..storage import LocalStorage, LocalStorageNode
from ..types import ASGIApp, Receive, Scope, Send
from ..config import AsgiConfigBuilder, ConfigurationHandler
from ..wsx.handler import WsxHandler
from genro_routes import RoutingClass # type: ignore[import-untyped]
__all__ = ["AsgiServer"]
[docs]
class AsgiServer(BasicAuthMixin, RoutingClass):
"""
ASGI server: a demultiplexer over the mounted applications.
The server does no routing of its own. It boots from a ``config.py`` whose
``ServerConfiguration`` builder is rendered onto it, mounting one app per
URL prefix in ``apps``; the Dispatcher selects the app from the first path
segment and the app does its own routing.
Attributes:
config_path: Path to the ``config.py`` that configures this server.
config_handler: ConfigurationHandler that renders the builder onto it.
dispatcher: Dispatcher; demultiplexes requests to the target app.
lifespan: ServerLifespan for startup/shutdown.
server_application: ServerApplication with system endpoints (``_server``).
apps: Dict of mounted apps keyed by mount (the empty mount is the main app).
wsx_handler: WsxHandler for WebSocket (WSX protocol) connections.
storage: LocalStorage service shared by the apps.
user_store: FileUserStore local identity source when ``users=True``, else None.
resource_loader: ResourceLoader with hierarchical app/server fallback.
session_store: Server-managed session store.
request_registry: RequestRegistry tracking the current request.
logger: Server logger instance.
"""
__slots__ = (
"_auth_config",
"config_path",
"configuration_class",
"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",
"user_store",
"login_enabled",
)
[docs]
def __init__(
self,
config_path: str | Path | type[AsgiConfigBuilder],
host: str | None = None,
port: int | None = None,
reload: bool | None = None,
parent: Any = None,
) -> None:
"""Initialize AsgiServer from a ``config.py`` path or a configuration class.
Composes BasicAuthMixin (auth) and RoutingClass (routing). Builds the
fixed parts of the server, then loads the configuration (a module with
a ``ServerConfiguration`` builder, or directly an ``AsgiConfigBuilder``
subclass) and renders it onto this server: the render is what mounts
apps, builds the middleware chain and plugs the router. The server is
the render target (see ``apply_configuration``).
Args:
config_path: Path to the ``config.py`` (its directory is the server
dir), or an ``AsgiConfigBuilder`` subclass for a programmatic
configuration (the server dir is the working directory).
host: Runtime override for the host (wins over the configuration).
port: Runtime override for the port (wins over the configuration).
reload: Runtime override for auto-reload (wins over the configuration).
parent: Optional parent object with shared resources and logic.
"""
self.parent = parent
self.db_registry: dict[str, Any] = {}
if isinstance(config_path, type) and issubclass(config_path, AsgiConfigBuilder):
self.configuration_class: type[AsgiConfigBuilder] | None = config_path
self.config_path = None
self.base_dir: Path = Path.cwd()
else:
self.configuration_class = None
self.config_path = Path(config_path).resolve()
self.base_dir = self.config_path.parent
self.host = host
self.port = port
self.reload = reload
self.openapi_info: dict[str, Any] = {}
self.apps: dict[str, AsgiApplication] = {}
# Fixed parts (one per instance, not configuration-driven).
BasicAuthMixin.__init__(self)
self.session_store = MemorySessionStore()
self.storage = LocalStorage(self.base_dir)
self.user_store: FileUserStore | None = None # set by _apply_server when users=True
self.login_enabled = False # set True by _apply_server when login=True (challenge on)
self.resource_loader = ResourceLoader(self)
self.logger = logging.getLogger("genro_asgi")
self.lifespan = ServerLifespan(self)
self.request_registry = RequestRegistry()
self.dispatcher: ASGIApp = Dispatcher(self)
self.wsx_handler = WsxHandler(self)
# Server application - system endpoints, mounted as the _server app.
self.server_application = ServerApplication(server=self)
self.server_application.mount_name = "_server"
self.apps["_server"] = self.server_application
# Configuration: read config.py, render the builder onto this server.
self.config_handler = ConfigurationHandler(self)
self.config_handler.configure(self._load_configuration())
def _load_configuration(self) -> AsgiConfigBuilder:
"""Instantiate the configuration builder (from the class or the config.py).
With a programmatic configuration (an ``AsgiConfigBuilder`` subclass
passed to the constructor) the class is instantiated directly; otherwise
the ``config.py`` module must define a ``ServerConfiguration`` subclass
of AsgiConfigBuilder — anything else is an error.
"""
if self.configuration_class is not None:
return self.configuration_class(name="config")
if not self.config_path.is_file():
raise FileNotFoundError(f"configuration file not found: {self.config_path}")
spec = importlib.util.spec_from_file_location("genro_asgi_config", self.config_path)
if spec is None or spec.loader is None:
raise ImportError(f"cannot load configuration module: {self.config_path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
configuration: type[AsgiConfigBuilder] | None = getattr(module, "ServerConfiguration", None)
if configuration is None:
raise AttributeError(
f"{self.config_path} does not define a 'ServerConfiguration' class"
)
return configuration(name="config")
# -- Configuration application (this server is the render target) --
[docs]
def apply_configuration(self, nodes: list[Any]) -> None:
"""Apply the rendered configuration nodes to this server.
Each top-level ConfigNode is dispatched by tag to ``_apply_<tag>``. An
unknown tag raises: grammar and server must stay in sync.
"""
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)
def _apply_server(self, node: Any) -> None:
"""Set host/port/reload, then attach the opt-in private sections.
A runtime override (constructor) wins for host/port/reload. The private
server sections are opt-in flags on the ``server`` node (default off);
they are not separate mounts — each is attached to the already-mounted
ServerApplication so its endpoints live under ``_server/<section>``.
The ``plugins`` flag attaches the plugin configurator UI under
``_server/plugins`` (the URL section — not ``src/genro_asgi/plugins/``,
which is the genro-routes plugin package). The ``users`` flag creates the
local identity store (``user_store``) and attaches the user-management
section under ``_server/users``; ``admin_password`` bootstraps the
built-in ``admin`` superadmin.
"""
if self.host is None:
self.host = node.attrs.get("host")
if self.port is None:
self.port = node.attrs.get("port")
if self.reload is None:
self.reload = node.attrs.get("reload")
if "storage_key" in node.attrs:
storage_key = node.attrs["storage_key"]
if not storage_key:
raise ValueError(
"server 'storage_key' is configured but resolved empty: "
"the storage encryption key is missing (check the environment)"
)
self.storage.set_encryption_keys(storage_key)
if node.attrs.get("plugins"):
section = PluginConfigSection(self.server_application)
self.server_application.attach_instance(section, name="plugins")
if node.attrs.get("users"):
self.user_store = FileUserStore(self.storage)
self.server_application.attach_section(
UsersSection(self.server_application), name="users"
)
if "admin_password" in node.attrs:
self._bootstrap_admin(node.attrs["admin_password"])
if node.attrs.get("login"):
self.login_enabled = True
self.server_application.register_auth_method(
PasswordMethod(self.server_application, "password")
)
def _bootstrap_admin(self, admin_password: str) -> None:
"""Create the built-in ``admin`` superadmin when the store has none.
``admin_password`` configured but resolved empty is an explicit boot
error (the ``storage_key`` pattern). When the store already holds an
enabled ``superadmin`` the password is ignored — the env var can stay
set; it never overwrites a managed store.
"""
if not admin_password:
raise ValueError(
"server 'admin_password' is configured but resolved empty: "
"the bootstrap admin password is missing (check the environment)"
)
store = self.user_store
has_superadmin = any(
record.get("enabled") and "superadmin" in record.get("tags", [])
for record in store.load_all()
)
if has_superadmin:
return
store.save(
{
"username": "admin",
"password_hash": store.hash_password(admin_password),
"tags": ["superadmin"],
"enabled": True,
"metadata": {"display_name": "", "email": ""},
}
)
def _apply_middleware(self, node: Any) -> None:
"""Build the global middleware chain and assign it to the dispatcher."""
self.dispatcher = middleware_chain(node.attrs, Dispatcher(self))
def _apply_authMiddleware(self, node: Any) -> None:
"""Configure authentication from bearer/basic/jwt attributes."""
BasicAuthMixin.__init__(self, **node.attrs)
def _apply_applications(self, node: Any) -> None:
"""Mount every application of the collection on its derived mount.
Each child's ``code`` is its collection key. The mount is the explicit
``mount`` attribute if given, else the empty mount ("") when the code
equals the collection ``default``, else the code itself. Every app does
its own routing through ``handle_request``; the server only
demultiplexes on the first path segment.
"""
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)
def _apply_oidc(self, node: Any) -> None:
"""Register one OIDC provider as an auth method (``oidc:<code>``).
Called once per ``root.oidc(...)`` node. ``code`` names the method
(``oidc:<code>``); the remaining attributes are the OAuth parameters. A
duplicate ``code`` is rejected downstream by the method registry (unique
method ids). Registering a provider also brings up the login page
machinery; ``login=True`` is what additionally arms the challenge.
"""
attrs = dict(node.attrs)
code = attrs.pop("code")
self.server_application.register_auth_method(
OidcMethod(self.server_application, code, **attrs)
)
def _apply_openapi(self, node: Any) -> None:
"""Set the OpenAPI metadata dict."""
self.openapi_info = node.attrs
def _apply_databases(self, node: Any) -> None:
"""Register each child ``database`` in db_registry, keyed by its ``code``.
Each database becomes a handler wrapping the db built by its
``db_class`` (see ``build_db_handler``).
"""
for child in node.children:
descriptor = dict(child.attrs)
code = descriptor.pop("code")
self.register_db(code, self.build_db_handler(code, descriptor))
[docs]
def build_db_handler(self, code: str, descriptor: dict[str, Any]) -> Any:
"""Build a db handler from a ``database`` descriptor (``code`` removed).
``db_class`` (required, imported by the config.py) builds the db from
the remaining attributes; ``db_handler_class`` (optional, default
``AsgiDbHandlerBase``) wraps it. The handler is what the registry holds.
"""
params = dict(descriptor)
db_class = params.pop("db_class", None)
if db_class is None:
raise ValueError(f"database {code!r}: 'db_class' is required")
db_handler_class = params.pop("db_handler_class", AsgiDbHandlerBase)
return db_handler_class(db_class(**params))
def _mount_app(self, node: Any, mount: str) -> None:
"""Instantiate one application node and register it under ``mount``.
``app_class`` is the application class (imported by the config.py);
``code`` and ``mount`` are consumed by the caller; the remaining
attributes are constructor kwargs. The app then interprets its own
configuration node (``databases`` and any tag of its grammar) in
``configure``: the server does not interpret an app's config.
"""
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)
[docs]
def create_session(self, auth: dict[str, Any] | None = None) -> Session:
"""Create a new session in the session store.
Args:
auth: Auth dict snapshot to store in session.
Returns:
New Session instance with unique token.
"""
return self.session_store.create(auth=auth)
[docs]
def mount(self, name: str, app: Any) -> None:
"""Mount an application on the server.
Args:
name: Mount name (becomes the URL prefix).
app: Application instance (AsgiApplication or RoutingClass).
"""
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
[docs]
def register_db(self, name: str, db: Any) -> None:
"""Register a database handler in the registry.
Args:
name: Logical name for this database (e.g. "sourcerer", "default").
db: A db handler (``AsgiDbHandlerBase``) wrapping the real db.
"""
self.db_registry[name] = db
[docs]
def get_db(self, name: str) -> Any:
"""Return a registered database handler by name.
Args:
name: Logical name used in register_db().
Raises:
KeyError: If no database registered with that name.
"""
return self.db_registry[name]
@property
def db(self) -> Any:
"""Default database handler (shortcut for get_db("default"))."""
return self.db_registry["default"]
# -- Plugin config persistence --
@property
def plugin_config_file(self) -> Path:
"""Plain path of plugin_config.json in server_dir (no-encryption mode)."""
return self.base_dir / "plugin_config.json"
@property
def plugin_config_node(self) -> LocalStorageNode:
"""Storage node for plugin_config.json.
Routes to the encrypted ``secure:`` mount when key material is
installed (storage.encryption_active); otherwise stays plain on the
``site:`` mount at the historical path.
"""
mount = "secure" if self.storage.encryption_active else "site"
return self.storage.node(f"{mount}:plugin_config.json")
[docs]
def get_plugin_config(self, router_path: str = "", entry: str = "") -> dict[str, Any]:
"""Read live plugin configuration for a router/entry.
Uses routing.configure("?") for introspection via genro-routes
public API.
Args:
router_path: Slash-separated path to the router (empty = all).
entry: Specific entry/handler name to filter (empty = all).
Returns:
Dict with router_path, plugin_info. Error dict if router not found.
"""
try:
description = self.routing.configure("?")
except (AttributeError, KeyError) as exc:
return {"error": str(exc)}
if router_path:
parts = router_path.split("/")
current = description
for part in parts:
routers = current.get("routers", {}) if isinstance(current, dict) else {}
if part not in routers:
return {"error": f"Router not found: {router_path}"}
current = routers[part]
description = {router_path.split("/")[-1]: current}
return {"router_path": router_path, "entry": entry, "plugin_info": description}
[docs]
def set_plugin_config(
self,
router_path: str = "",
plugin_name: str = "",
target: str = "_all_",
**config_values: Any,
) -> dict[str, Any]:
"""Apply a plugin config change at runtime without persisting.
Delegates to routing.configure() via genro-routes public API.
Args:
router_path: Slash-separated path to the router (empty = root).
plugin_name: Name of the plugin to configure.
target: Target slot ("_all_" or specific entry name).
**config_values: Plugin configuration key-value pairs.
Returns:
Dict with "ok": True on success, or "error" key on failure.
"""
if not plugin_name:
return {"error": "plugin_name is required"}
router_spec = router_path.replace("/", "/") if router_path else "root"
selector = f"/{target}" if target != "_all_" else ""
config_target = f"{router_spec}:{plugin_name}{selector}"
try:
self.routing.configure(config_target, **config_values)
except (AttributeError, KeyError, ValueError) as exc:
return {"error": str(exc)}
return {"ok": True}
[docs]
def save_plugin_config(self) -> dict[str, Any]:
"""Persist current runtime plugin config to plugin_config.json.
Uses routing.configure("?") to snapshot the full tree via
genro-routes public API. Writes through plugin_config_node: encrypted
at rest when keys are installed, plain otherwise (with a warning).
Returns:
Dict with "ok": True and "file" fullpath on success.
"""
snapshot = self.routing.configure("?")
node = self.plugin_config_node
if not self.storage.encryption_active:
self.logger.warning(
"storage_key not configured: plugin config persisted unencrypted at %s",
self.plugin_config_file,
)
node.write_text(json.dumps(snapshot, indent=2))
return {"ok": True, "file": node.fullpath}
[docs]
def revert_plugin_config(self) -> dict[str, Any]:
"""Revert runtime config to what is saved in plugin_config.json.
Returns:
Dict with "ok": True. Includes "message" if no saved config exists.
"""
data = self._load_persisted_config()
if not data:
return {"ok": True, "message": "No saved config, nothing to revert"}
self._apply_config_data(data)
return {"ok": True}
[docs]
def apply_persisted_config(self) -> None:
"""Apply plugin_config.json to the current router tree.
Called at boot after plugins are attached.
"""
data = self._load_persisted_config()
if data:
self._apply_config_data(data)
def _load_persisted_config(self) -> dict[str, Any]:
"""Load plugin_config.json if it exists (through plugin_config_node)."""
node = self.plugin_config_node
if node.exists:
result: dict[str, Any] = json.loads(node.read_text())
return result
return {}
def _apply_config_data(self, data: dict[str, Any]) -> None:
"""Apply config data dict to the live router tree.
Delegates to routing.configure() for each entry.
"""
for router_name, router_info in data.items():
for plugin_info in router_info.get("plugins", []):
plugin_name = plugin_info.get("name", "")
config = plugin_info.get("config", {})
if not config:
continue
config_target = f"{router_name}:{plugin_name}"
try:
self.routing.configure(config_target, **config)
except (AttributeError, KeyError, ValueError) as exc:
self.logger.warning(
"plugin_config.json: cannot apply %s: %s", config_target, exc
)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
Handle an ASGI event.
Routes by scope type: lifespan to self.lifespan, everything else
(http and websocket) to the Dispatcher through the middleware chain.
The Dispatcher demultiplexes on the first path segment; websocket
scopes fall back to the server's WsxHandler when the mounted app
exposes no websocket hook.
Args:
scope: ASGI scope dict.
receive: ASGI receive callable.
send: ASGI send callable.
"""
if scope["type"] == "lifespan":
await self.lifespan(scope, receive, send)
else:
await self.dispatcher(scope, receive, send)
[docs]
def run(self, workers: int | None = None) -> None:
"""Run the server using Uvicorn.
``workers`` > 1 runs that many uvicorn processes on one shared socket
(each rebuilds the server through the module factory, like reload does);
mutually exclusive with reload — uvicorn's own constraint.
"""
host = self.host or "127.0.0.1"
port = self.port or 8000
reload = self.reload or False
if reload and workers and workers > 1:
raise ValueError("reload and workers are mutually exclusive (uvicorn constraint)")
self.logger.info(f"Starting server on {host}:{port}")
if reload or (workers and workers > 1):
# Both modes need an import string and a fresh instance per process/
# restart: hand the configuration to the module factory via env —
# the config path, or the dotted configuration class.
self.export_factory_env()
uvicorn.run(
"genro_asgi.server.server:server_factory",
host=host,
port=port,
reload=reload,
reload_dirs=[str(self.base_dir)] if reload else None,
workers=workers if not reload else None,
factory=True,
)
else:
uvicorn.run(self, host=host, port=port)
[docs]
def export_factory_env(self) -> None:
"""Expose this server's configuration to ``server_factory`` via env.
A file configuration travels as its path (GENRO_ASGI_CONFIG); a
programmatic one as the dotted ``module:Class`` of the configuration
class (GENRO_ASGI_CONFIG_CLASS), which must therefore be importable.
"""
if self.configuration_class is not None:
cls = self.configuration_class
os.environ.pop("GENRO_ASGI_CONFIG", None)
os.environ["GENRO_ASGI_CONFIG_CLASS"] = f"{cls.__module__}:{cls.__qualname__}"
else:
os.environ.pop("GENRO_ASGI_CONFIG_CLASS", None)
os.environ["GENRO_ASGI_CONFIG"] = str(self.config_path)
def __repr__(self) -> str:
"""Return string representation."""
apps_str = ", ".join(f"{path!r}" for path in self.apps)
return f"AsgiServer(apps=[{apps_str}])"
@property
def request(self) -> BaseRequest | None:
"""Current request from registry."""
return self.request_registry.current
@property
def response(self) -> Response | None:
"""Current response from request."""
req = self.request
return req.response if req else None
def server_factory() -> AsgiServer:
"""Build an AsgiServer from the configuration handed over via env.
Used by uvicorn's reload and multi-worker modes, which need an import
string and a fresh instance per restart/process. ``GENRO_ASGI_CONFIG``
carries a config.py path; ``GENRO_ASGI_CONFIG_CLASS`` a dotted
``module:Class`` of an ``AsgiConfigBuilder`` subclass (the programmatic
configuration, e.g. the quickstart one reading its own env parameters).
"""
dotted = os.environ.get("GENRO_ASGI_CONFIG_CLASS")
if dotted:
module_path, _, class_name = dotted.partition(":")
module = importlib.import_module(module_path)
return AsgiServer(getattr(module, class_name))
return AsgiServer(os.environ["GENRO_ASGI_CONFIG"])