orchestrata_ultimate.plugin.loader

Reference › orchestrata_ultimate › plugin › loader

PluginLoader finds plugin candidates, imports their classes and orders them by dependency. The PluginManager owns one, and discovery calls it.

Discovery has two sources. discover_local scans each configured plugin directory (with ~ expanded) for subdirectories that hold a plugin.yaml, parses each with PluginManifest.from_yaml from manifest, records the directory as the manifest's source path and skips, with a warning, any invalid manifest. discover_installed reads the orchestrata_plugin entry-point group through importlib.metadata, loads each entry point, and when the class has get_manifest instantiates it and takes the manifest; a failing entry point is logged and skipped. discover_all combines both.

load_plugin_class(manifest) expects an entry_point of the form module.path:ClassName, calls _validate_module_path, imports the module and returns the class without instantiating it. _validate_module_path raises PluginLoadError when the dotted path does not start with one of the allowed prefixes or contains a blocked segment such as os or subprocess. Import errors and a missing class are also PluginLoadError. topological_sort orders manifests so dependencies come first; the plugin narrative records that a cycle raises PluginCircularDependencyError and a missing dependency raises PluginDependencyError.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PluginLoader

Decorators: traces(spec='015', story='US-001', reqs=['REQ-015-006']) · Line 46

Discovers and loads plugin modules from filesystem and pip.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/plugin/loader.py

"""PluginLoader — discovers and loads plugins from disk and pip.

Handles two discovery sources:
1. Local directories — scans for plugin.yaml files
2. Pip-installed packages — queries importlib.metadata for the
   orchestrata_plugin entry point group

Also handles module import and class instantiation.
"""

import importlib
import importlib.metadata
import logging
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.exceptions import (
    PluginDependencyError,
    PluginLoadError,
    PluginManifestError,
)
from orchestrata_ultimate.plugin.manifest import PluginManifest

_LOG = logging.getLogger(__name__)

ALLOWED_PLUGIN_PREFIXES: tuple[str, ...] = (
    "orchestrata_ultimate.plugins.",
    "orchestrata_plugin_",
)

_BLOCKED_MODULE_SEGMENTS: frozenset[str] = frozenset(
    {
        "os",
        "subprocess",
        "shutil",
        "importlib",
        "ctypes",
        "multiprocessing",
        "signal",
    }
)


@traces(spec="015", story="US-001", reqs=["REQ-015-006"])
class PluginLoader:
    """Discovers and loads plugin modules from filesystem and pip.

    Supports two discovery sources:
    - Local directories containing plugin.yaml manifests
    - Pip-installed packages with orchestrata_plugin entry points
    """

    def __init__(
        self,
        plugin_dirs: list[str] | None = None,
        allowed_prefixes: tuple[str, ...] | None = None,
    ) -> None:
        """Initialize the plugin loader.

        Args:
            plugin_dirs: Directories to scan for local plugins.
                Defaults to ["plugins"].
            allowed_prefixes: Module path prefixes allowed for
                plugin entry points. Defaults to
                ALLOWED_PLUGIN_PREFIXES.
        """
        self._plugin_dirs = plugin_dirs or ["plugins"]
        self._allowed_prefixes = (
            allowed_prefixes
            if allowed_prefixes is not None
            else ALLOWED_PLUGIN_PREFIXES
        )

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-006"],
    )
    def discover_local(self) -> list[PluginManifest]:
        """Discover plugins from local directories.

        Scans each configured plugin directory for subdirectories
        containing a plugin.yaml file.

        Returns:
            List of validated plugin manifests found locally.
        """
        manifests: list[PluginManifest] = []
        for dir_path_str in self._plugin_dirs:
            dir_path = Path(dir_path_str).expanduser()
            if not dir_path.exists():
                _LOG.debug(
                    "Plugin directory does not exist: %s",
                    dir_path,
                )
                continue
            for child in sorted(dir_path.iterdir()):
                if not child.is_dir():
                    continue
                manifest_path = child / "plugin.yaml"
                if not manifest_path.exists():
                    continue
                try:
                    manifest = PluginManifest.from_yaml(manifest_path)
                    manifest.source_path = child
                    manifests.append(manifest)
                    _LOG.info(
                        "Discovered local plugin: %s v%s at %s",
                        manifest.name,
                        manifest.version,
                        child,
                    )
                except PluginManifestError as exc:
                    _LOG.warning(
                        "Skipping invalid manifest at %s: %s",
                        manifest_path,
                        exc,
                    )
        return manifests

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-007"],
    )
    def discover_installed(self) -> list[PluginManifest]:
        """Discover plugins from pip-installed packages.

        Queries importlib.metadata for packages in the
        orchestrata_plugin entry point group.

        Returns:
            List of manifests from installed plugin packages.
        """
        manifests: list[PluginManifest] = []
        try:
            entry_points = importlib.metadata.entry_points(group="orchestrata_plugin")
        except Exception:
            _LOG.debug("No orchestrata_plugin entry points found")
            return manifests

        for ep in entry_points:
            try:
                plugin_cls = ep.load()
                if hasattr(plugin_cls, "get_manifest"):
                    instance = plugin_cls()
                    manifest = instance.get_manifest()
                    manifests.append(manifest)
                    _LOG.info(
                        "Discovered installed plugin: %s v%s",
                        manifest.name,
                        manifest.version,
                    )
            except Exception as exc:
                _LOG.warning(
                    "Failed to load entry point '%s': %s",
                    ep.name,
                    exc,
                )
        return manifests

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-009"],
    )
    def discover_all(self) -> list[PluginManifest]:
        """Discover plugins from all sources, deduplicated.

        Local plugins take precedence over installed ones when
        names collide.

        Returns:
            Deduplicated list of plugin manifests.
        """
        local = self.discover_local()
        installed = self.discover_installed()
        seen: set[str] = set()
        result: list[PluginManifest] = []
        for manifest in local:
            if manifest.name not in seen:
                seen.add(manifest.name)
                result.append(manifest)
        for manifest in installed:
            if manifest.name not in seen:
                seen.add(manifest.name)
                result.append(manifest)
        _LOG.info(
            "Discovered %d plugins total (%d local, %d installed)",
            len(result),
            len(local),
            len(installed),
        )
        return result

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-011"],
    )
    def _validate_module_path(
        self,
        module_path: str,
        plugin_name: str,
    ) -> None:
        """Validate that a module path is allowed for import.

        Checks that the module path starts with an allowed
        prefix and does not contain blocked module segments
        (e.g., os, subprocess).

        Args:
            module_path: Dotted Python module path.
            plugin_name: Plugin name for error context.

        Raises:
            PluginLoadError: If the module path is not allowed.
        """
        if not module_path.startswith(self._allowed_prefixes):
            raise PluginLoadError(
                f"Module path '{module_path}' for plugin "
                f"'{plugin_name}' does not start with an "
                f"allowed prefix: {self._allowed_prefixes}",
                context={
                    "plugin": plugin_name,
                    "module": module_path,
                    "allowed_prefixes": list(
                        self._allowed_prefixes,
                    ),
                },
            )
        segments = set(module_path.split("."))
        blocked = segments & _BLOCKED_MODULE_SEGMENTS
        if blocked:
            raise PluginLoadError(
                f"Module path '{module_path}' for plugin "
                f"'{plugin_name}' contains blocked segments: "
                f"{sorted(blocked)}",
                context={
                    "plugin": plugin_name,
                    "module": module_path,
                    "blocked_segments": sorted(blocked),
                },
            )

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-011"],
    )
    def load_plugin_class(self, manifest: PluginManifest) -> Any:
        """Import and return the plugin class from the manifest.

        The entry_point field is a dotted Python path in the form
        "package.module:ClassName".

        Args:
            manifest: Plugin manifest with entry_point field.

        Returns:
            The plugin class (not instantiated).

        Raises:
            PluginLoadError: If the module or class cannot be
                imported.
        """
        entry = manifest.entry_point
        if ":" not in entry:
            raise PluginLoadError(
                f"Invalid entry_point format '{entry}': "
                f"expected 'module.path:ClassName'",
                context={
                    "plugin": manifest.name,
                    "entry_point": entry,
                },
            )
        module_path, class_name = entry.rsplit(":", 1)
        self._validate_module_path(module_path, manifest.name)
        try:
            module = importlib.import_module(module_path)
        except ImportError as exc:
            raise PluginLoadError(
                f"Cannot import module '{module_path}' for "
                f"plugin '{manifest.name}': {exc}",
                context={
                    "plugin": manifest.name,
                    "module": module_path,
                },
            ) from exc

        cls = getattr(module, class_name, None)
        if cls is None:
            raise PluginLoadError(
                f"Class '{class_name}' not found in module "
                f"'{module_path}' for plugin '{manifest.name}'",
                context={
                    "plugin": manifest.name,
                    "module": module_path,
                    "class": class_name,
                },
            )
        return cls

    @traces(spec="015", story="US-001", reqs=["REQ-015-014"])
    @staticmethod
    def topological_sort(
        manifests: list[PluginManifest],
    ) -> list[PluginManifest]:
        """Delegate to module-level topological_sort.

        Args:
            manifests: List of plugin manifests to sort.

        Returns:
            Manifests sorted so dependencies come first.
        """
        return topological_sort(manifests)


@traces(
    spec="015",
    story="US-010",
    reqs=["REQ-015-132"],
)
def topological_sort(
    manifests: list[PluginManifest],
) -> list[PluginManifest]:
    """Sort manifests in dependency order (topological sort).

    Args:
        manifests: List of plugin manifests to sort.

    Returns:
        Manifests sorted so dependencies come first.

    Raises:
        PluginDependencyError: If a dependency is not found
            among the provided manifests.
    """
    from orchestrata_ultimate.plugin.exceptions import (
        PluginCircularDependencyError,
    )

    name_map = {m.name: m for m in manifests}
    visited: set[str] = set()
    in_stack: set[str] = set()
    result: list[PluginManifest] = []

    @traces(spec="015", story="US-001", reqs=["REQ-015-015"])
    def visit(name: str) -> None:
        if name in visited:
            return
        if name in in_stack:
            raise PluginCircularDependencyError(
                f"Circular dependency detected involving plugin '{name}'",
                context={"plugin": name},
            )
        in_stack.add(name)
        manifest = name_map.get(name)
        if manifest is None:
            raise PluginDependencyError(
                f"Dependency '{name}' not found among discovered plugins",
                context={"missing": name},
            )
        for dep in manifest.dependencies:
            visit(dep)
        in_stack.discard(name)
        visited.add(name)
        result.append(manifest)

    for m in manifests:
        visit(m.name)
    return result