src/genro_asgi/plugin_mixin.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Plugin capability: router plugins as a mixin over the base server (D16).16 17 The base server knows nothing about router plugins. This mixin adds them as a18 capability, composed BEFORE the server class (``class MyServer(PluginMixin,19 BaseServer)``), mirroring ``MiddlewareMixin``: its cooperative ``__init__``20 peels ``plugins=`` (the ``{name: bool | dict}`` switches — a dict value enables21 the plugin and becomes its plug options) and ``plugin_registry=`` (extra22 ``{name: class}`` entries merged over ``default_plugin_registry()``), and23 exposes ``arm_router(router)``.24 25 Unlike the middleware chain — assembled once around the base dispatch —26 plugins are armed onto the ROUTER of each routed application: ``arm_router``27 is called by ``RoutedApplication`` on first ``route`` access. Per enabled28 plugin it ensures the plugin class is registered with genro-routes (idempotent),29 then batch-plugs the enabled set with a single ``router.plug([...])`` call30 (genro-routes 0.28.0 accepts a list of plugin dicts); bundled genro-routes31 plugins (auth/channel/env/logging/pydantic) carry no class in the registry and32 are plugged by name alone. A composition WITHOUT this mixin exposes no33 ``arm_router`` and arms nothing — ``RoutedApplication`` degrades silently.34 35 ``default_plugin_registry()`` returns a FRESH dict per call (``{"openapi":36 OpenAPIPlugin}`` as of Phase 5) — a function, so no module-level mutable37 registry exists; and importing this module never registers a plugin against38 genro-routes (no import side effect).39 """40 41 from __future__ import annotations42 43 from typing import TYPE_CHECKING, Any44 45 from genro_routes import Router46 47 from .plugins.openapi import OpenAPIPlugin48 49 if TYPE_CHECKING:50 from genro_routes.plugins._base_plugin import BasePlugin51 52 __all__ = ["PluginMixin", "default_plugin_registry"]53 54 55 #: The plugins every server arms unconditionally — the fixed structure of a56 #: routed core, not a config choice. ``pydantic`` captures handler signatures57 #: into the neutral ``params``/``result`` blocks; ``openapi`` carries the58 #: per-entry schema controls (``openapi_method`` and friends). The ``plugins``59 #: config section only ADDS extras over this base; disabling a fixed plugin is60 #: a config error, not an opt-out.61 FIXED_PLUGINS: tuple[str, ...] = ("pydantic", "openapi")62 63 64 def default_plugin_registry() -> dict[str, type[BasePlugin]]:65 """A fresh ``{name: class}`` mapping of the plugins this core arms itself.66 67 Only the transport-dialect plugins that live in this package need a class68 here (so ``arm_router`` can register them). genro-routes' own bundled69 plugins are plugged by name without a class.70 """71 return {"openapi": OpenAPIPlugin}72 73 74 class PluginMixin:75 """Router-plugin capability mixin, composed BEFORE a server class.76 77 Every server arms the ``FIXED_PLUGINS`` (``pydantic``/``openapi``)78 unconditionally — the fixed structure of a routed core. Constructor kwargs79 peeled here: ``plugins`` — the ``{name: bool | dict}`` switches tuning the80 base and adding extras (a dict value becomes the plug options; ``False``81 drops an EXTRA but is a config error on a fixed plugin);82 ``plugin_registry`` — extra ``{name: class}`` entries merged over83 ``default_plugin_registry()``.84 """85 86 def __init__(self, **kwargs: Any) -> None:87 plugins: dict[str, bool | dict[str, Any]] | None = kwargs.pop("plugins", None)88 extra: dict[str, type[BasePlugin]] | None = kwargs.pop("plugin_registry", None)89 super().__init__(**kwargs)90 registry = default_plugin_registry()91 if extra:92 registry.update(extra)93 self._plugin_registry = registry94 self._plugins_config = self._resolve_plugins(plugins or {})95 96 def _resolve_plugins(97 self, config: dict[str, bool | dict[str, Any]]98 ) -> dict[str, dict[str, Any]]:99 """Merge the config switches over the fixed base into ``{name: options}``.100 101 The result always carries the ``FIXED_PLUGINS`` (``pydantic``/``openapi``)102 — the fixed structure of a routed core. The ``{name: bool | dict}``103 config only tunes the base and adds extras: a dict value is the plug104 options; a truthy scalar enables an extra with no options;105 ``False``/``None`` drops an EXTRA. Disabling a fixed plugin106 (``{"openapi": False}``) is a config error, not an opt-out.107 """108 resolved: dict[str, dict[str, Any]] = {name: {} for name in FIXED_PLUGINS}109 for name, value in config.items():110 if not value and name in FIXED_PLUGINS:111 raise ValueError(f"Plugin '{name}' is fixed structure and cannot be disabled")112 if isinstance(value, dict):113 resolved[name] = value114 elif value:115 resolved[name] = {}116 else:117 resolved.pop(name, None)118 return resolved119 120 @property121 def plugins(self) -> dict[str, dict[str, Any]]:122 """The materialized ``{name: options}`` config of the enabled plugins."""123 return dict(self._plugins_config)124 125 @property126 def plugin_registry(self) -> dict[str, type[BasePlugin]]:127 """The effective ``{name: class}`` registry (defaults + extras)."""128 return dict(self._plugin_registry)129 130 def arm_router(self, router: Router) -> None:131 """Register and plug every enabled plugin onto ``router`` (idempotent).132 133 A plugin carrying a class in the registry is registered with134 genro-routes first (guarded — never re-registered); the not-yet-attached135 plugins are then armed with a single batch ``router.plug([...])`` call, so136 arming twice is safe (already-attached plugins are skipped). Unknown137 plugin names surface as ``router.plug`` errors, not silent no-ops.138 """139 plugged = {plugin.name for plugin in router.iter_plugins()}140 specs: list[dict[str, Any]] = []141 for name, options in self._plugins_config.items():142 cls = self._plugin_registry.get(name)143 if cls is not None and name not in Router.available_plugins():144 Router.register_plugin(cls)145 if name not in plugged:146 specs.append({"name": name, **options})147 if specs:148 router.plug(specs)