# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Base class for mountable ASGI applications.
AsgiApplication provides a default router, resource loading, and lifecycle
hooks (on_init, on_startup, on_shutdown). Subclass it to build apps that the
server mounts on a URL prefix (the first path segment) from a ``config.py``
recipe via the ``root.applications()`` collection (``apps.application(
code=..., mount=...)``).
"""
from __future__ import annotations
import asyncio
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, ClassVar
from genro_routes import RoutingClass, is_result_wrapper, route # type: ignore[import-untyped]
from genro_toolbox import DictObj # type: ignore[import-untyped]
from genro_toolbox.smartasync import smartasync # type: ignore[import-untyped]
from ...exceptions import ROUTER_ERRORS, HTTPNotFound
from ...middleware import middleware_chain
from ...request import set_current_request
if TYPE_CHECKING:
from ...server import AsgiServer
from ...types import Receive, Scope, Send
__all__ = ["AsgiApplication"]
RESOURCES_DIR = Path(__file__).parent / "resources"
_INDEX_HTML = (RESOURCES_DIR / "index.html").read_text()
[docs]
class AsgiApplication(RoutingClass):
"""Base class for apps mounted on AsgiServer.
Routes live on the app's single router (the lazy ``route`` property) and
provide a default `index()` method. Subclasses define `openapi_info` for
metadata and add routes with the @route() decorator; sub-trees are built
by attaching child RoutingClass instances.
Example::
class MyApp(AsgiApplication):
openapi_info = {"title": "My API", "version": "1.0.0"}
@route()
def hello(self):
return "Hello!"
def on_init(self, **kwargs):
# Sub-tree under /backoffice via a child instance
self.attach_instance(Backoffice(self), name="backoffice")
"""
app_protocol: ClassVar[str] = "asgi"
openapi_info: ClassVar[dict[str, Any]] = {}
default_db_name: ClassVar[str | None] = None
# Channel this app's HTTP face presents to the router. The path already tells
# which face serves a request (REST here, a distinct segment for MCP), so the
# face declares its own channel — no client header needed.
rest_channel: ClassVar[str] = "rest"
[docs]
def __init__(self, **kwargs: Any) -> None:
"""Set up mount name, app path, db registry and middleware slot, then call on_init.
Args:
**kwargs: ``base_dir`` and ``db_name`` are consumed here; the rest is
forwarded to ``on_init()``.
"""
self._mount_name: str = ""
self.base_dir = kwargs.pop("base_dir", None)
self._app_path: Path = Path(self.base_dir) if self.base_dir else self._default_app_path()
self.db_name: str | None = kwargs.pop("db_name", None) or self.default_db_name
self.db_registry: dict[str, Any] = {}
self._chain: Any = None
self.on_init(**kwargs)
def _default_app_path(self) -> Path:
"""Directory of the module that defines this app's class.
Every app lives where its source lives, so by default an app finds its
own files (resources, templates) next to its class. A ``base_dir`` kwarg
or the server (via the ``app_path`` setter at mount time) can override it.
"""
module = sys.modules.get(type(self).__module__)
module_file = getattr(module, "__file__", None)
return Path(module_file).parent if module_file else Path.cwd()
@property
def mount_name(self) -> str:
"""Return the name under which this app is mounted on the server."""
return self._mount_name
@mount_name.setter
def mount_name(self, value: str) -> None:
self._mount_name = value
@property
def app_path(self) -> Path:
"""Filesystem directory where this app is located.
Defaults to the directory of the module that defines the app's class;
the server may override it at mount time. Resources, templates and other
per-app files are resolved relative to this path.
"""
return self._app_path
@app_path.setter
def app_path(self, value: str | Path) -> None:
self._app_path = Path(value)
[docs]
def on_init(self, **kwargs: Any) -> None:
"""Called after base initialization. Override for custom setup.
Args:
**kwargs: Constructor kwargs from the ``app``/``application`` entry
in the config.py recipe.
"""
pass
[docs]
def make_callable(self, node: Any, request: Any) -> Callable[[], Any]:
"""Build the complete call this app wants to execute for ``node``.
Not just the kwargs: the whole call, arguments AND lifecycle — packaged
in the vehicle the handler needs. The node declares its handler's nature
(genro-routes marks async entries so ``iscoroutinefunction(node)`` is
honest); the caller runs the result via ``smartasync(...)()`` either way:
- sync handler -> sync zero-arg callable -> executor thread, which is
where thread-local resource teardown belongs (a thread-local db
connection must be closed on the same thread it opened, which the
request-level cleanup on the loop cannot do);
- async handler -> async zero-arg callable awaited on the event loop,
with ``route_cleanup`` running there AFTER the handler — an async
handler must not lean on thread-local resources.
Two hooks vary the behaviour, this wrapper does not: ``bind_kwargs``
decides the arguments (REST spread, etc.); ``route_cleanup`` runs after
the handler. Both default to a no-op-ish base; subclasses/mixins
override the one they own.
"""
kwargs = self.bind_kwargs(node, request)
# asyncio.iscoroutinefunction, not inspect's: on 3.11 genro-routes falls
# back to the asyncio sentinel, which only the asyncio check reads.
if asyncio.iscoroutinefunction(node):
async def acall() -> Any:
try:
return await node(**kwargs)
finally:
self.route_cleanup()
return acall
def call() -> Any:
try:
return node(**kwargs)
finally:
self.route_cleanup()
return call
[docs]
def bind_kwargs(self, node: Any, request: Any) -> dict[str, Any]:
"""Return the kwargs to call ``node`` with. Base: the request kwargs.
Subclasses with a wire dialect (REST/JSON) override to reconcile payload
conventions with the handler signature (see OpenApiApplication).
"""
return request.handler_kwargs()
[docs]
def route_cleanup(self) -> None:
"""Teardown after the handler, in the executor thread. Base: nothing.
A subclass/mixin that owns a resource (e.g. a legacy db connection)
overrides this to release it here — where it is thread-correct, unlike
the request-level cleanup that runs on the loop.
"""
pass
[docs]
def spread_over_params(self, node: Any, data: dict[str, Any]) -> dict[str, Any]:
"""Spread a dict of values over the handler's declared parameters.
Shared by every wire dialect (REST body, MCP arguments): a structured
payload is fitted to a scalar-parameter handler. 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 means a known no-parameter
handler: everything is dropped. Reads the neutral ``params`` block.
"""
fields = node.params.get("fields")
if fields is None: # no signature description at all: keep whole
return data
accepts_kwargs = any(f["kind"] == "var_keyword" for f in fields)
if accepts_kwargs:
return data
param_names = {
f["name"] for f in fields if f["kind"] not in ("var_positional", "var_keyword")
}
return {k: v for k, v in data.items() if k in param_names}
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Entry point seen by the dispatcher: the app as an ASGI callable.
With a middleware chain the request flows through it (the chain ends at
``handle_request``); without one it goes straight to ``handle_request``.
This is the app's own onion layer of the relay.
"""
if self._chain is not None:
await self._chain(scope, receive, send)
else:
await self.handle_request(scope, receive, send)
@property
def server(self) -> AsgiServer | None:
"""Return the server that mounted this app (semantic alias for _routing_parent)."""
return getattr(self, "_routing_parent", None)
[docs]
def get_db(self, name: str) -> Any:
"""Return a db handler by name: the app registry first, then the server.
Returns ``None`` when no database is registered under ``name`` in either
registry (the app's own databases shadow the server's shared ones).
"""
if name in self.db_registry:
return self.db_registry[name]
server = self.server
if server is not None and name in server.db_registry:
return server.get_db(name)
return None
@property
def db(self) -> Any:
"""Default db handler for this app (resolves ``db_name`` or "default")."""
return self.get_db(self.db_name or "default")
[docs]
def path_in_app(self, path: str) -> str:
"""Return the request path relative to this app's mount.
The server demultiplexes on the first path segment; an app resolves
the remainder in its own router. The empty mount keeps the whole path.
"""
if not self.mount_name:
return path
prefix = f"/{self.mount_name}"
rest = path[len(prefix) :] if path.startswith(prefix) else path
return rest or "/"
[docs]
def authenticate(self, scope: Scope) -> dict[str, Any] | None:
"""Resolve the identity for a request. Default: delegate to the server.
Override to give an app its own authentication. Returning the server's
auth keeps the current shared behaviour.
"""
server = self.server
return server.authenticate(scope) if server is not None else None
[docs]
def monitor_state(self) -> dict[str, Any]:
"""This app's contribution to the server monitor: identity facts.
The server-level monitor aggregates one entry per mounted app by calling
this on each. Subclasses override to add their specific panel data
(registries, pools, gauges) on top of these base facts.
"""
return {
"class": type(self).__name__,
"protocol": self.app_protocol,
"mount": self.mount_name,
}
[docs]
def monitor_descriptor(self) -> dict[str, Any]:
"""How this app's panel LOOKS on the server monitor: a static descriptor.
The presentation complement of ``monitor_state()``: the state carries the
data (polled), the descriptor tells the monitor shell which panel renders
it (fetched once — it is a per-class constant). ``panel`` names a renderer
in the shell's registry; this default (and any name the shell does not
know) falls back to the generic panel — raw state as key/value rows and
tables. Subclasses override to declare their specific panel and the extra
endpoints it needs (see ``SpaMultiWorkerApplication``).
"""
return {"panel": "generic"}
[docs]
async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Resolve the request in this app's router and send the response.
This is the per-app routing engine: authenticate, resolve the handler
in ``self.route`` from the path relative to the mount, build the context,
execute and respond. Subclasses with a foreign protocol (e.g. a WSGI
proxy) override this entirely.
"""
server = self.server
if server is None:
raise RuntimeError("handle_request requires a mounted app")
auth = self.authenticate(scope)
scope["auth"] = auth
filters: dict[str, Any] = {
"auth_tags": auth["tags"] if auth is not None else [],
"channel_channel": self.rest_channel,
}
request = await server.request_registry.create(scope, receive, send, server=server)
set_current_request(request)
error: BaseException | None = None
try:
node = self.route.node(self.path_in_app(request.path), errors=ROUTER_ERRORS, **filters)
self.on_route_resolved(node, request)
ctx = DictObj()
ctx.server = server
ctx.app = self
ctx.request = request
ctx.session = scope.get("session")
ctx.avatar = auth.get("avatar") if auth else None
db = self.get_db(self.db_name or "default")
if db is not None:
ctx._db = db
scope["ctx"] = ctx
result = await smartasync(self.make_callable(node, request))()
if is_result_wrapper(result):
metadata: dict[str, Any] = {**node.metadata, **result.metadata}
request.response.set_result(result.value, metadata)
else:
request.response.set_result(result, node.metadata)
await request.response(scope, receive, send)
except BaseException as exc:
error = exc
raise
finally:
request.run_cleanups(error=error)
set_current_request(None)
server.request_registry.unregister()
[docs]
def on_route_resolved(self, node: Any, request: Any) -> None:
"""Hook called right after route resolution, before the handler runs.
A no-op seam on the base (FIXED): subclasses observe which route a request
resolved to — e.g. a pool worker stamps per-connection activity here for
routes not marked as system traffic (``@route(meta_sysrpc=True)``).
"""
pass
[docs]
def on_startup(self) -> None:
"""Called when server starts. Override for custom initialization.
Can be sync or async. Called after all apps are mounted.
"""
pass
[docs]
def on_shutdown(self) -> None:
"""Called when server stops. Override for custom cleanup.
Can be sync or async. Called in reverse order of startup.
"""
pass
@property
def resources_dir(self) -> Path | None:
"""Return path to app's resources directory.
Derived from ``app_path`` (the app's own location), so an app finds its
resources whether mounted on the server or used standalone.
"""
path = self.app_path / "resources"
return path if path.is_dir() else None
[docs]
@route()
def static(self, file: str = "") -> Path:
"""Serve a static file from the app's resources directory.
Returns Path - set_result() handles mime type detection.
Raises:
HTTPNotFound: If no file is named, or the resource does not exist.
"""
if not file:
raise HTTPNotFound("static: file name required")
file_path = self.app_path / "resources" / file
if not file_path.exists() or not file_path.is_file():
raise HTTPNotFound(f"Resource not found: {file}")
return file_path
[docs]
def load_resource(self, *args: str, name: str) -> tuple[bytes, str] | None:
"""Load a resource file as ``(content_bytes, mime_type)`` or ``None``.
Mounted: tries the server's ResourceLoader first, then falls back to the
app's own ``resources_dir``. Standalone: reads ``resources_dir`` directly,
with a small suffix-based mime map (octet-stream default).
"""
if self.server:
mount_name = self.mount_name
result = self.server.resource_loader.load(mount_name, *args, name=name)
if result:
return result
# Fallback to the app's own resources_dir (under app_path).
# Local mode: read from resources_dir
if not self.resources_dir:
return None
resource_path = (
self.resources_dir / "/".join(args) / name if args else self.resources_dir / name
)
if resource_path.exists():
# Simple mime type detection
suffix = resource_path.suffix.lower()
mime_types = {
".html": "text/html",
".css": "text/css",
".js": "application/javascript",
".json": "application/json",
".txt": "text/plain",
}
mime_type = mime_types.get(suffix, "application/octet-stream")
return resource_path.read_bytes(), mime_type
return None
[docs]
@route(media_type="text/html")
def index(self) -> str:
"""Return HTML splash page. Override for custom index."""
info = getattr(self, "openapi_info", {})
title = info.get("title", self.__class__.__name__)
version = info.get("version", "")
description = info.get("description", "")
version_html = f"<p>Version: {version}</p>" if version else ""
desc_html = f"<p>{description}</p>" if description else ""
return _INDEX_HTML.format(
title=title,
version_html=version_html,
desc_html=desc_html,
)
class _AppHandler:
"""Innermost ASGI app of an application's own middleware chain.
The chain ends here: this adapter turns the chain tail into a call to the
application's ``handle_request``. It is built with the app it serves, so it
needs no per-request lookup (unlike a server-held chain).
"""
__slots__ = ("application",)
def __init__(self, application: AsgiApplication) -> None:
self.application = application
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.application.handle_request(scope, receive, send)
if __name__ == "__main__":
pass