# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""OpenApiApplication - mounts a RoutingClass with automatic OpenAPI + docs."""
from __future__ import annotations
import importlib
from typing import Any, ClassVar
from genro_routes import RoutingClass, route # type: ignore[import-untyped]
from pathlib import Path
from genro_asgi.plugins.openapi import router_openapi
from ..asgi_application import AsgiApplication
__all__ = ["OpenApiApplication"]
RESOURCES_DIR = Path(__file__).parent / "resources"
_SWAGGER_HTML = (RESOURCES_DIR / "swagger.html").read_text()
_OPENAPI_INDEX_HTML = (RESOURCES_DIR / "openapi_index.html").read_text()
[docs]
class OpenApiApplication(AsgiApplication):
"""Wrap a RoutingClass (or the app's own @route methods) as REST + OpenAPI.
Adds meta endpoints under ``_meta/`` (schema, docs, splash) alongside the API:
- /{app}/_meta/schema_json — OpenAPI 3.1 JSON schema
- /{app}/_meta/docs — Swagger UI page
- /{app}/_meta/index — splash page
Two ways to supply the API:
- mounted mode: pass ``routing_class=`` (or ``module=``); the class is
attached under ``api_name`` at /{app}/{api_name}/endpoint.
- direct mode: subclass and write ``@route`` methods on the app itself;
endpoints sit at the app root /{app}/endpoint.
Example (mounted)::
app = OpenApiApplication(routing_class=api)
server.mount("sourcerer", app)
Example (direct)::
class ShopApi(OpenApiApplication):
@route(media_type="application/json")
def items(self) -> dict: ...
"""
openapi_info: ClassVar[dict[str, Any]] = {}
[docs]
def on_init(
self,
routing_class: RoutingClass | None = None,
module: str | None = None,
docs: str = "swagger",
api_name: str = "api",
**kwargs: Any,
) -> None:
"""Initialize with a routing class instance or import path.
Args:
routing_class: RoutingClass instance to mount.
module: Import path "package.module:ClassName" (alternative to routing_class).
docs: Documentation style — "swagger", "redoc", or "off".
api_name: Path segment the routing class is attached under.
**kwargs: Passed to the routing class constructor if module is used.
"""
self._docs_style = docs
self._api_name = api_name
self._openapi_info: dict[str, Any] = {}
self.attach_instance(OpenApiMeta(self), name="_meta")
if routing_class is None and module:
routing_class = self._import_routing_class(module, **kwargs)
if routing_class is not None:
self._mount_routing_class(routing_class)
[docs]
def bind_kwargs(self, node: Any, request: Any) -> dict[str, Any]:
"""Reconcile a JSON body with a REST handler's scalar parameters.
A JSON body arrives as a single ``body_data`` dict (request.py). A
classic REST handler declares scalar parameters, not ``body_data``, so
the dict must be spread over the fields the handler accepts. The handler
signature comes from the node's neutral ``params`` block (never inspect).
The whole ``body_data`` is kept when the handler itself declares
``body_data`` or accepts ``**kwargs`` — it wants the raw object or any
key. Otherwise the dict is spread over the declared names, extras dropped.
"""
kwargs = request.handler_kwargs()
body = kwargs.get("body_data")
if not isinstance(body, dict):
return kwargs
fields = node.params.get("fields")
if fields is None: # no signature (no pydantic plugin): keep body_data whole
return kwargs
param_names = {
f["name"] for f in fields if f["kind"] not in ("var_positional", "var_keyword")
}
accepts_kwargs = any(f["kind"] == "var_keyword" for f in fields)
if "body_data" in param_names or accepts_kwargs: # handler absorbs the raw object
return kwargs
del kwargs["body_data"]
kwargs.update(self.spread_over_params(node, body))
return kwargs
@property
def docs_style(self) -> str:
"""Documentation style — "swagger", "redoc", or "off"."""
return self._docs_style
@property
def api_name(self) -> str:
"""Path segment the routing class is attached under."""
return self._api_name
@property
def api_info(self) -> dict[str, Any]:
"""OpenAPI info dict: the mounted class's, else the app's, else empty."""
return self._openapi_info or self.openapi_info or {}
def _import_routing_class(self, module_path: str, **kwargs: Any) -> Any:
"""Import and instantiate a RoutingClass from an import path.
Args:
module_path: "package.module:ClassName" format.
**kwargs: Constructor arguments.
Returns:
Instantiated RoutingClass.
"""
module_name, class_name = module_path.split(":")
mod = importlib.import_module(module_name)
cls = getattr(mod, class_name)
return cls(**kwargs)
def _mount_routing_class(self, routing_class: RoutingClass) -> None:
"""Mount the routing class under ``api_name`` on this app's router.
The pydantic plugin is plugged on the mounted class's own router: it is
what captures the handler signatures into the neutral ``params``/``result``
blocks that OpenAPI, MCP and ``make_callable`` all rely on. Plugging
an already-bound router re-decorates its existing entries, so the class
need not know about it. (Later this becomes app-configuration-driven.)
Args:
routing_class: The RoutingClass instance to mount.
"""
self.attach_instance(routing_class, name=self.api_name)
routing_class.route.plug("pydantic")
info = getattr(routing_class, "openapi_info", None)
if info:
self._openapi_info = info
class OpenApiMeta(RoutingClass):
"""Meta endpoints of an OpenApiApplication, attached under ``_meta``."""
def __init__(self, application: OpenApiApplication) -> None:
self.application = application
@route()
def schema_json(self) -> dict[str, Any]:
"""Return OpenAPI 3.1 JSON schema for mounted routes.
Delegates to ServerApplication.openapi() when mounted on a server,
falls back to local generation for standalone usage.
"""
app = self.application
mount_name = app.mount_name
if app.server and mount_name:
result: dict[str, Any] = app.server.server_application.openapi(mount_name)
return result
# Standalone fallback: same REST-channel traversal the server path uses.
paths_data = router_openapi(app.route, channel_channel=app.rest_channel)
info = app.api_info
return {
"openapi": "3.1.0",
"info": {
"title": info.get("title", type(app).__name__),
"version": info.get("version", "1.0.0"),
"description": info.get("description", ""),
},
**paths_data,
}
@route(media_type="text/html")
def docs(self) -> str:
"""Serve Swagger UI page pointing to the server's openapi endpoint."""
app = self.application
if app.docs_style == "off":
return "Documentation disabled."
mount_name = app.mount_name
schema_url = f"/_server/openapi/{mount_name}" if mount_name else "/_server/openapi"
title = app.api_info.get("title", "API")
return _SWAGGER_HTML.format(title=title, schema_url=schema_url)
@route(media_type="text/html")
def index(self) -> str:
"""Splash page with link to docs."""
app = self.application
info = app.api_info
title = info.get("title", type(app).__name__)
version = info.get("version", "")
description = info.get("description", "")
mount_name = app.mount_name
docs_url = f"/{mount_name}/_meta/docs" if mount_name else "/_meta/docs"
version_html = f"<p>Version: {version}</p>" if version else ""
desc_html = f"<p>{description}</p>" if description else ""
return _OPENAPI_INDEX_HTML.format(
title=title,
version_html=version_html,
desc_html=desc_html,
docs_url=docs_url,
)
if __name__ == "__main__":
pass