"""Dynamic plugin registration into an :class:`ActionRegistry`.
``PackageLoader`` imports an external package by name and registers every
top-level function / class / builtin under the key ``"<package>_<member>"``.
"""
from __future__ import annotations
from importlib import import_module
from importlib.util import find_spec
from inspect import getmembers, isbuiltin, isclass, isfunction
from types import ModuleType
from automation_file.core.action_registry import ActionRegistry
from automation_file.logging_config import file_automation_logger
[docs]
class PackageLoader:
"""Load packages lazily and register their public callables."""
def __init__(self, registry: ActionRegistry) -> None:
self.registry: ActionRegistry = registry
self._cache: dict[str, ModuleType] = {}
[docs]
def load(self, package: str) -> ModuleType | None:
"""Import ``package`` once and return the module (cached)."""
cached = self._cache.get(package)
if cached is not None:
return cached
spec = find_spec(package)
if spec is None:
file_automation_logger.error("PackageLoader: cannot find %s", package)
return None
try:
# `package` is a trusted caller-supplied name (see PackageLoader docstring and
# the CLAUDE.md security note on plugin loading); it is not untrusted input.
name = spec.name
module = import_module(name) # nosemgrep
except ImportError as error:
file_automation_logger.error("PackageLoader import error: %r", error)
return None
self._cache[package] = module
return module
[docs]
def add_package_to_executor(self, package: str) -> int:
"""Register every function / class / builtin from ``package``.
Returns the number of commands that were registered.
"""
module = self.load(package)
if module is None:
return 0
count = 0
for predicate in (isfunction, isbuiltin, isclass):
for member_name, member in getmembers(module, predicate):
self.registry.register(f"{package}_{member_name}", member)
count += 1
file_automation_logger.info("PackageLoader: registered %d members from %s", count, package)
return count