# 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 URL prefixes via config.yaml.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar
from genro_routes import Router, RoutingClass, route # type: ignore[import-untyped]
if TYPE_CHECKING:
from ...server import AsgiServer
__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.
Provides default `main` router and `index()` method. Subclasses define
`openapi_info` for metadata and add routes with @route() decorator.
Example::
class MyApp(AsgiApplication):
openapi_info = {"title": "My API", "version": "1.0.0"}
@route() # Uses the only router (self.main) automatically
def hello(self):
return "Hello!"
def __init__(self, **kwargs):
super().__init__(**kwargs) # Required: creates self.main
self.backoffice = Router(self, name="backoffice")
@route("backoffice") # Must specify when multiple routers
def admin(self):
return "Admin panel"
"""
app_protocol: ClassVar[str] = "asgi"
openapi_info: ClassVar[dict[str, Any]] = {}
default_db_name: ClassVar[str | None] = None
[docs]
def __init__(self, **kwargs: Any) -> None:
"""Initialize app with default main router.
Args:
**kwargs: Passed to ``on_init()`` after extracting ``base_dir``
and ``db_name``.
"""
self._mount_name: str = ""
self.base_dir = kwargs.pop("base_dir", None)
self.db_name: str | None = kwargs.pop("db_name", None) or self.default_db_name
self.main = Router(self, name="main")
self.on_init(**kwargs)
@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
[docs]
def on_init(self, **kwargs: Any) -> None:
"""Called after base initialization. Override for custom setup.
Args:
**kwargs: Parameters from config.yaml app definition.
"""
pass
@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 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.
Works both when mounted on server and standalone.
"""
if self.base_dir:
return Path(self.base_dir) / "resources"
return None
[docs]
def load_resource(self, *args: str, name: str) -> tuple[bytes, str] | None:
"""Load resource file content with mime type.
When mounted on server: uses server's ResourceLoader with fallback to local.
Standalone: reads directly from resources_dir.
Returns:
Tuple of (content_bytes, mime_type) or None if not found.
"""
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 local resources_dir (for sys_apps under _sys/)
# 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(meta_mime_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,
)