"""Router with plugin pipeline for Genro Routes.
``Router`` extends ``BaseRouter`` with a global plugin registry, per-router
plugin instances, middleware wrapping, and plugin state stored on the router.
Internal state
--------------
- ``_plugin_specs``: list of ``_PluginSpec`` (factory, kwargs copy).
- ``_plugins``: instantiated plugins in the order they were attached.
- ``_plugins_by_name``: name → plugin instance (first wins).
- ``_inherited_from``: set of parent ids already inherited to avoid double
cloning when the same child is attached multiple times.
- ``_plugin_info``: per-plugin state store on the router.
Global registry
---------------
``Router.register_plugin(name, plugin_class)`` validates that ``plugin_class``
is a subclass of ``BasePlugin`` and ``name`` is non-empty.
Attaching plugins
-----------------
``plug(plugin_name, **config)`` looks up the plugin class by name in the global
registry. It stores a ``_PluginSpec``, instantiates the plugin, appends to
``_plugins`` and ``_plugins_by_name``, applies ``plugin.on_decore`` to all
existing entries, rebuilds handlers, and returns ``self``.
Wrapping pipeline
-----------------
``_wrap_handler(entry, call_next)`` builds middleware layers from the current
``_plugins`` in reverse order (last attached closest to the handler).
Inheritance behaviour
---------------------
``_on_attached_to_parent(parent)`` runs when a child router is attached.
Parent specs are cloned once per parent. Cloned specs are instantiated into
new plugins that are prepended ahead of existing child plugins.
Example::
from genro_routes import Router, RoutingClass, route
class MyService(RoutingClass):
def __init__(self):
self.route.plug("logging")
@route()
def hello(self):
return "Hello!"
"""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from functools import wraps
from typing import Any
from genro_toolbox import dictExtract
from genro_routes.core.base_router import BaseRouter
from genro_routes.plugins._base_plugin import BasePlugin, MethodEntry
__all__ = ["Router"]
_PLUGIN_REGISTRY: dict[str, type[BasePlugin]] = {}
@dataclass
class _PluginSpec:
"""Specification for creating plugin instances."""
factory: type[BasePlugin]
kwargs: dict[str, Any]
def instantiate(self, router: Router) -> BasePlugin:
"""Create a plugin instance for the given router."""
return self.factory(router=router, **self.kwargs)
def clone(self) -> _PluginSpec:
"""Return a copy with shallow-copied kwargs."""
return _PluginSpec(self.factory, dict(self.kwargs))
[docs]
class Router(BaseRouter):
"""Router with plugin registry and pipeline support.
Extends BaseRouter with:
- Global plugin registry for registering plugin classes
- Per-router plugin instances with middleware wrapping
- Plugin state management and configuration
- Plugin inheritance when attaching child routers
"""
__slots__ = BaseRouter.__slots__ + (
"_plugin_specs",
"_plugins",
"_plugins_by_name",
"_inherited_from",
"_plugin_info",
"_plugin_children",
)
[docs]
def __init__(self, *args, **kwargs):
self._plugin_specs: list[_PluginSpec] = []
self._plugins: list[BasePlugin] = []
self._plugins_by_name: dict[str, BasePlugin] = {}
self._inherited_from: set[int] = set()
self._plugin_info: dict[str, dict[str, Any]] = {}
self._plugin_children: dict[str, list[Router]] = {} # plugin_name -> [child routers]
super().__init__(*args, **kwargs)
# ------------------------------------------------------------------
# Plugin registration
# ------------------------------------------------------------------
[docs]
@classmethod
def register_plugin(cls, plugin_class: type[BasePlugin], name: str | None = None) -> None:
"""Register a plugin class globally.
Args:
plugin_class: A BasePlugin subclass with plugin_code defined.
name: Optional override name. If provided, overwrites any existing
registration. If not provided, uses plugin_code and raises
if already registered with a different class.
Raises:
TypeError: If plugin_class is not a BasePlugin subclass.
ValueError: If plugin_code is missing or name collision occurs.
"""
if not isinstance(plugin_class, type) or not issubclass(plugin_class, BasePlugin):
raise TypeError("plugin_class must be a BasePlugin subclass")
if not getattr(plugin_class, "plugin_code", None):
raise ValueError(
f"Plugin {plugin_class.__name__} not following standards: missing plugin_code"
)
code = name or plugin_class.plugin_code
# If name is explicitly provided, allow overwrite (intentional replacement)
# Otherwise, reject collision
if name is None:
existing = _PLUGIN_REGISTRY.get(code)
if existing is not None and existing is not plugin_class:
raise ValueError(f"Plugin '{code}' already registered")
_PLUGIN_REGISTRY[code] = plugin_class
[docs]
@classmethod
def available_plugins(cls) -> dict[str, type[BasePlugin]]:
"""Return a copy of the global plugin registry."""
return dict(_PLUGIN_REGISTRY)
[docs]
def plug(self, plugin: str | list[dict[str, Any]], **config: Any) -> Router:
"""Attach one or more plugins by name (previously registered globally).
Args:
plugin: Either a plugin name (single attach) or a list of plugin
dicts ``{"name": ..., <options>}`` (batch attach). Each dict
carries the plugin name plus its own options; shared kwargs are
not allowed with a list.
**config: Configuration options passed to the plugin (single form only).
Returns:
self (for method chaining).
Raises:
TypeError: If plugin is not a string or a list of dicts.
ValueError: If a plugin is not registered, already attached, a list
is mixed with shared kwargs, or a dict lacks 'name'.
"""
if isinstance(plugin, (list, tuple)):
if config:
raise ValueError("Do not mix shared kwargs with a list of plugins")
for item in plugin:
if not isinstance(item, dict):
raise TypeError(
f"Each plugin spec must be a dict, got {type(item).__name__}"
)
options = dict(item)
try:
name = options.pop("name")
except KeyError as err:
raise ValueError("Plugin dict must include 'name'") from err
self.plug(name, **options)
return self
if not isinstance(plugin, str):
raise TypeError(
f"Plugin must be referenced by name string, got {type(plugin).__name__}"
)
plugin_class = _PLUGIN_REGISTRY.get(plugin)
if plugin_class is None:
available = ", ".join(sorted(_PLUGIN_REGISTRY)) or "none"
raise ValueError(
f"Unknown plugin '{plugin}'. Register it first. Available plugins: {available}"
)
if plugin in self._plugins_by_name:
raise ValueError(
f"Plugin '{plugin}' is already attached to this router. "
"Use configure() to update settings."
)
spec_kwargs = dict(config)
spec = _PluginSpec(plugin_class, spec_kwargs)
self._plugin_specs.append(spec)
instance = spec.instantiate(self)
self._plugins.append(instance)
self._plugins_by_name[instance.name] = instance
# Plugin will be applied to entries during lazy binding (_bind)
# If already bound, apply now
if self._bound:
self._apply_plugin_to_entries(instance)
self._rebuild_handlers()
# Propagate to already-attached children so plug() is order-independent
# (plug-then-attach and attach-then-plug produce the same result).
self._propagate_plugin_to_children(instance)
return self
[docs]
def iter_plugins(self) -> list[BasePlugin]: # type: ignore[override]
"""Return attached plugin instances in application order."""
return list(self._plugins)
[docs]
def get_config(self, plugin_name: str, method_name: str | None = None) -> dict[str, Any]:
"""Return merged plugin configuration.
Retrieves the effective configuration for a plugin, merging global
(router-level) settings with optional per-handler overrides.
Args:
plugin_name: Name of the attached plugin.
method_name: If provided, includes handler-specific overrides
merged on top of global config.
Returns:
Dict of configuration values. Per-handler values override global.
Raises:
AttributeError: If no plugin with that name is attached.
Example:
>>> router.plug("logging")
>>> router.logging.configure(before=False)
>>> router.logging.configure(_target="slow_handler", after=True)
>>> router.get_config("logging")
{'enabled': True, 'before': False}
>>> router.get_config("logging", "slow_handler")
{'enabled': True, 'before': False, 'after': True}
"""
plugin = self._plugins_by_name.get(plugin_name)
if plugin is None:
raise AttributeError(
f"No plugin named '{plugin_name}' attached to router '{self.name}'"
)
return plugin.configuration(method_name)
[docs]
def __getattr__(self, name: str) -> Any:
"""Access attached plugins by name as attributes.
Enables fluent plugin configuration via attribute access syntax.
Only works for attached plugins; other attribute access follows
normal Python behavior.
Args:
name: Plugin name (e.g., "logging", "auth", "pydantic").
Returns:
The BasePlugin instance attached under that name.
Raises:
AttributeError: If no plugin with that name is attached.
Example:
>>> router.plug("logging").plug("auth")
>>> router.logging.configure(before=False)
>>> router.auth.configure(rule="admin")
"""
plugin = self._plugins_by_name.get(name)
if plugin is None:
raise AttributeError(f"No plugin named '{name}' attached to router '{self.name}'")
return plugin
def _get_plugin_bucket(self, plugin_name: str) -> dict[str, Any] | None:
"""Get the plugin_info bucket for a plugin, initializing _all_ if needed."""
bucket = self._plugin_info.get(plugin_name)
if bucket is not None and "_all_" not in bucket:
bucket["_all_"] = {"config": {}, "locals": {}}
return bucket
# ------------------------------------------------------------------
# Runtime helpers (state stored on plugin_info)
# ------------------------------------------------------------------
[docs]
def set_plugin_enabled(self, method_name: str, plugin_name: str, enabled: bool = True) -> None:
"""Enable or disable a plugin for a specific handler at runtime.
Sets a runtime override in the "locals" store, which takes precedence
over static configuration. Use "_all_" as method_name to affect all
handlers globally.
Args:
method_name: Handler name or "_all_" for global setting.
plugin_name: Name of the attached plugin.
enabled: True to enable, False to disable.
Raises:
AttributeError: If no plugin with that name is attached.
Example:
>>> router.set_plugin_enabled("slow_handler", "logging", False)
>>> router.set_plugin_enabled("_all_", "pydantic", False)
"""
bucket = self._get_plugin_bucket(plugin_name)
if bucket is None:
raise AttributeError(
f"No plugin named '{plugin_name}' attached to router '{self.name}'"
)
entry = bucket.setdefault(method_name, {"config": {}, "locals": {}})
entry.setdefault("locals", {})["enabled"] = bool(enabled)
[docs]
def is_plugin_enabled(self, method_name: str, plugin_name: str) -> bool:
"""Check if a plugin is enabled for a specific handler.
Resolution order (first found wins):
1. entry locals (runtime override via set_plugin_enabled)
2. entry config (static via configure(_target=method_name, enabled=...))
3. global locals (runtime override via set_plugin_enabled for _all_)
4. global config (static via configure(enabled=...))
5. default: True
"""
bucket = self._get_plugin_bucket(plugin_name)
if bucket is None:
raise AttributeError(
f"No plugin named '{plugin_name}' attached to router '{self.name}'"
)
# Check entry-level first
entry_data = bucket.get(method_name, {})
if "enabled" in entry_data.get("locals", {}):
return bool(entry_data["locals"]["enabled"])
if "enabled" in entry_data.get("config", {}):
return bool(entry_data["config"]["enabled"])
# Then check global (_all_)
base_data = bucket.get("_all_", {})
if "enabled" in base_data.get("locals", {}):
return bool(base_data["locals"]["enabled"])
if "enabled" in base_data.get("config", {}):
return bool(base_data["config"]["enabled"])
return True
[docs]
def set_runtime_data(self, method_name: str, plugin_name: str, key: str, value: Any) -> None:
"""Store arbitrary runtime data for a plugin/handler combination.
Plugins can use this to store handler-specific state that persists
across invocations but is not part of the configuration schema.
Args:
method_name: Handler name or "_all_" for global data.
plugin_name: Name of the attached plugin.
key: Data key to store.
value: Value to store (any type).
Raises:
AttributeError: If no plugin with that name is attached.
Example:
>>> router.set_runtime_data("handler", "auth", "last_access", time.time())
"""
bucket = self._get_plugin_bucket(plugin_name)
if bucket is None:
raise AttributeError(
f"No plugin named '{plugin_name}' attached to router '{self.name}'"
)
entry = bucket.setdefault(method_name, {"config": {}, "locals": {}})
entry.setdefault("locals", {})[key] = value
[docs]
def get_runtime_data(
self, method_name: str, plugin_name: str, key: str, default: Any = None
) -> Any:
"""Retrieve runtime data for a plugin/handler combination.
Args:
method_name: Handler name or "_all_" for global data.
plugin_name: Name of the attached plugin.
key: Data key to retrieve.
default: Value to return if key not found.
Returns:
The stored value, or default if not found.
Raises:
AttributeError: If no plugin with that name is attached.
Example:
>>> last = router.get_runtime_data("handler", "auth", "last_access")
"""
bucket = self._get_plugin_bucket(plugin_name)
if bucket is None:
raise AttributeError(
f"No plugin named '{plugin_name}' attached to router '{self.name}'"
)
entry_locals = bucket.get(method_name, {}).get("locals", {})
return entry_locals.get(key, default)
# ------------------------------------------------------------------
# Overrides/hooks
# ------------------------------------------------------------------
def _wrap_handler(self, entry: MethodEntry, call_next: Callable) -> Callable: # type: ignore[override]
"""Build the middleware pipeline for a handler.
Wraps the handler with each plugin's wrap_handler in reverse order
(last plugin attached is closest to the handler).
Args:
entry: The MethodEntry being wrapped.
call_next: The base callable (entry.func).
Returns:
The fully wrapped handler with all plugin middleware applied.
"""
wrapped = call_next
for plugin in reversed(self._plugins):
plugin_call = plugin.wrap_handler(self, entry, wrapped)
wrapped = self._create_wrapper(plugin, entry, plugin_call, wrapped)
return wrapped
def _create_wrapper(
self,
plugin: BasePlugin,
entry: MethodEntry,
plugin_call: Callable,
next_handler: Callable,
) -> Callable:
"""Create a wrapper that checks plugin enabled state before invoking.
Args:
plugin: The plugin providing the wrapper.
entry: The entry being wrapped.
plugin_call: The plugin's wrapped callable.
next_handler: The next handler in the chain (to use if disabled).
Returns:
A wrapper that skips the plugin if disabled.
"""
@wraps(next_handler)
def wrapper(*args, **kwargs):
if not self.is_plugin_enabled(entry.name, plugin.name):
return next_handler(*args, **kwargs)
return plugin_call(*args, **kwargs)
return wrapper
def _apply_plugin_to_entries(self, plugin: BasePlugin) -> None:
"""Apply a plugin's on_decore to all existing entries.
Called when a plugin is attached after entries have been registered.
"""
# Access raw dict to avoid triggering lazy binding
for entry in self._BaseRouter__entries_raw.values(): # type: ignore[attr-defined]
if entry.router is not self:
continue # alias — belongs to source router
if plugin.name not in entry.plugins:
entry.plugins.append(plugin.name)
plugin.on_decore(self, entry.func, entry)
def _propagate_plugin_to_children(self, plugin: BasePlugin) -> None:
"""Propagate a single just-added plugin to already-attached children.
Makes plug() order-independent: a plugin added after children were
attached reaches those children (and their children, recursively), the
same way _on_attached_to_parent would have at attach time. Does not go
through _on_attached_to_parent, so the per-parent _inherited_from guard
is untouched. Idempotent: a child that already has the plugin is not
duplicated.
"""
for child in self._children.values():
if not isinstance(child, Router):
continue
owner = child.instance
# Only true (primary) children inherit; secondary links are just
# navigational shortcuts and keep their own parent's plugins.
if owner is None or getattr(owner, "_routing_parent", None) is not self.instance:
continue
registered = self._plugin_children.setdefault(plugin.name, [])
if child not in registered:
registered.append(child)
child_plugin = child._plugins_by_name.get(plugin.name)
if child_plugin is None:
child_plugin = plugin.__class__(child)
child._plugins_by_name[plugin.name] = child_plugin
child._plugins.append(child_plugin)
child_plugin.on_attached_to_parent(plugin)
# Apply to the child's own entries only if it is already bound;
# otherwise its lazy _bind() will run on_decore for every plugin.
if child._bound:
child._apply_plugin_to_entries(child_plugin)
child._rebuild_handlers()
# Recurse so grandchildren receive the plugin too.
child._propagate_plugin_to_children(child_plugin)
def _on_attached_to_parent(self, parent: Router) -> None: # type: ignore[override]
"""Handle plugin inheritance when this router is attached to a parent.
Creates or notifies child plugins based on parent's plugin configuration.
Inherited plugins receive on_attached_to_parent hook and on_decore is
called for all existing entries.
"""
parent_id = id(parent)
if parent_id in self._inherited_from:
return
self._inherited_from.add(parent_id)
inherited_plugins = []
for parent_plugin in parent._plugins:
if parent_plugin.name not in self._plugins_by_name:
# Child doesn't have this plugin - create new instance and inherit
child_plugin = parent_plugin.__class__(self)
# Register child in parent's notification list
parent._plugin_children.setdefault(parent_plugin.name, []).append(self)
# Add to child's plugin registry
self._plugins_by_name[parent_plugin.name] = child_plugin
self._plugins.append(child_plugin)
inherited_plugins.append((child_plugin, parent_plugin))
else:
# Child already has this plugin - let it handle inheritance
child_plugin = self._plugins_by_name[parent_plugin.name]
# Register for notifications even if child has its own plugin
parent._plugin_children.setdefault(parent_plugin.name, []).append(self)
# Call hook to let plugin decide what to do
child_plugin.on_attached_to_parent(parent_plugin)
# For inherited plugins: call hook and apply on_decore
for child_plugin, parent_plugin in inherited_plugins:
child_plugin.on_attached_to_parent(parent_plugin)
for entry in self._entries.values():
if child_plugin.name not in entry.plugins:
entry.plugins.append(child_plugin.name)
child_plugin.on_decore(self, entry.func, entry)
if inherited_plugins:
self._rebuild_handlers()
def _after_entry_registered(self, entry: MethodEntry) -> None: # type: ignore[override]
"""Process plugin config and apply on_decore after a handler is registered.
Extracts plugin_config from entry metadata and applies it via
plugin.configure(). Then calls on_decore for each attached plugin.
"""
for pname, cfg in entry.metadata.get("plugin_config", {}).items():
plugin = self._plugins_by_name.get(pname)
if plugin:
plugin.configure(_target=entry.name, **cfg)
else:
bucket = self._plugin_info.setdefault(
pname, {"_all_": {"config": {}, "locals": {}}}
)
bucket.setdefault(entry.name, {"config": {}, "locals": {}})["config"].update(cfg)
for plugin in self._plugins:
if plugin.name not in entry.plugins:
entry.plugins.append(plugin.name)
plugin.on_decore(self, entry.func, entry)
def _entry_invalid_reason(self, entry: MethodEntry | None, **allowing_args: Any) -> str:
"""Check if an entry should be denied based on plugin rules.
Consults each plugin's deny_reason() method to determine if access
should be denied. Plugin kwargs are extracted by prefix (e.g., auth_*
goes to auth plugin).
Args:
entry: The entry to check (None if not found).
**allowing_args: Plugin-prefixed filter kwargs.
Returns:
Empty string if allowed, otherwise the deny reason code.
"""
if entry is None:
return "not_found"
# Filter out None and False values
allowing_args = {k: v for k, v in allowing_args.items() if v not in (None, False)}
for plugin in self._plugins:
# Extract kwargs for this specific plugin using its plugin_code prefix
plugin_kwargs = dictExtract(
allowing_args, f"{plugin.plugin_code}_", slice_prefix=True, pop=False
)
# Always consult plugin - it decides based on entry rules and user kwargs
result = plugin.deny_reason(entry, **plugin_kwargs)
if result:
return result
return ""
def _describe_entry_extra( # type: ignore[override]
self, entry: MethodEntry, base_description: dict[str, Any]
) -> dict[str, Any]:
"""Gather plugin config, plugin metadata and the neutral result block."""
extra: dict[str, Any] = {}
plugins_info: dict[str, dict[str, Any]] = {}
for plugin in self._plugins:
plugin_data: dict[str, Any] = {}
# Get config for this entry
config = plugin.configuration(entry.name)
if config:
plugin_data["config"] = config
# Get metadata from plugin
meta = plugin.entry_metadata(self, entry)
if meta:
if not isinstance(meta, dict):
raise TypeError( # pragma: no cover - defensive guard
f"Plugin {plugin.name} returned non-dict "
f"from entry_metadata: {type(meta)}"
)
plugin_data["metadata"] = meta
# Only include plugin if it has data
if plugin_data:
plugins_info[plugin.name] = plugin_data
if plugins_info:
extra["plugins"] = plugins_info
result = self._describe_result(entry)
if result:
extra["result"] = result
params = self._describe_params(entry)
if params:
extra["params"] = params
return extra
def _describe_result(self, entry: MethodEntry) -> dict[str, Any]:
"""Build the dialect-neutral result block for a handler.
Combines the JSON Schema of the return type (produced by the pydantic
plugin at decoration time) with the declared result media type. The
block is present only when at least one is available; OpenAPI and MCP
translators read it instead of re-deriving the output description.
Returns:
``{"schema": <json schema | None>, "media_type": <str>}`` when a
return schema and/or a media type exists, else an empty dict.
"""
schema = entry.metadata.get("pydantic", {}).get("response_schema")
media_type = entry.metadata.get("meta", {}).get("media_type")
if schema is None and media_type is None:
return {}
return {"schema": schema, "media_type": media_type}
def _describe_params(self, entry: MethodEntry) -> dict[str, Any]:
"""Build the dialect-neutral input-params block for a handler.
Twin of the result block, for the input side: exposes the aggregate
input JSON Schema plus a per-parameter list. ``fields`` is the complete
parameter description in declaration order, each entry being
``{name, schema, required, default, kind}``; the var-parameters are
included with kind ``var_positional`` (``*args``) and ``var_keyword``
(``**kwargs``), so a consumer knows the handler accepts arbitrary
arguments from ``fields`` alone. Both are cached by the pydantic plugin
at decoration time; this only reads them, so nodes() never
re-serializes a schema. Present only when the pydantic plugin captured
params for this entry.
Returns:
``{"schema": <json schema | None>, "fields": [...]}`` when the
plugin captured params, else an empty dict.
"""
pydantic_meta = entry.metadata.get("pydantic", {})
if "param_fields" not in pydantic_meta:
return {}
return {
"schema": pydantic_meta.get("request_schema"),
"fields": pydantic_meta["param_fields"],
}