orchestrata_ultimate.plugin.sdk

Reference › orchestrata_ultimate › plugin › sdk

PluginSDK is the public facade over the PluginManager for operators and integrators who use the plugin system from code. It is also the composition root for plugin-registered Volks: its constructor creates a concrete VolkRegistry and passes it to the manager, so a plugin's register_volk call lands in a queryable registry and not nowhere. It takes an optional PluginConfig and an optional DI container, and exposes the manager as manager.

The methods are thin async or sync delegations: discover, load, unload, reload, enable, disable, install, uninstall, start_all and stop_all, plus the queries info, list_all and active_plugins, which return PluginInfo records. The install docstring lists local path, pip spec or git URL as sources; the manager's installer in discovery supports a local directory and a git source, and tells the caller to add pip-distributed plugins with the package manager so that entry-point discovery finds them. Errors are those of the manager, from exceptions.

At a glance

Class diagram

Dependency diagram

API

Classes

PluginSDK

Decorators: traces(spec='015', story='US-002', reqs=['REQ-015-017']) · Line 25

Public API surface for the Orchestrata Plugin System.

Source

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

"""PluginSDK — public API surface for plugin authors.

Provides a convenient facade over the plugin system for plugin
developers. Wraps PluginManager with simplified methods for
common operations like discovering, loading, and querying plugins.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.config import PluginConfig
from orchestrata_ultimate.plugin.interfaces import (
    PluginInfo,
    PluginStatus,
)
from orchestrata_ultimate.plugin.manager import PluginManager
from orchestrata_ultimate.plugin.manifest import PluginManifest
from orchestrata_ultimate.plugin.registries import VolkRegistry

_LOG = logging.getLogger(__name__)


@traces(spec="015", story="US-002", reqs=["REQ-015-017"])
class PluginSDK:
    """Public API surface for the Orchestrata Plugin System.

    Provides a simplified facade for interacting with the plugin
    manager, discovery, loading, and querying plugin state.
    Intended as the primary entry point for operators and
    integrators using the plugin system programmatically.

    Attributes:
        manager: The underlying PluginManager instance.
        volk_registry: The concrete VolkRegistry every plugin's
            ``register_volk`` call is wired into (SPEC-015 US-015-002).
    """

    @traces(
        spec="015",
        story="US-002",
        reqs=["REQ-015-017"],
    )
    def __init__(
        self,
        config: PluginConfig | None = None,
        di_container: Any = None,
    ) -> None:
        """Initialize the Plugin SDK.

        Args:
            config: Plugin manager configuration.
            di_container: DI container for plugin contexts.
        """
        # PluginSDK is the composition root for operators/integrators using
        # the plugin system programmatically, so it constructs the concrete
        # VolkRegistry a loaded plugin's ``register_volk`` calls land in
        # (previously PluginManager's ``volk_registry`` defaulted to None
        # here, making every plugin-registered Volk a silent no-op).
        self.volk_registry = VolkRegistry()
        self._manager = PluginManager(
            config=config,
            di_container=di_container,
            volk_registry=self.volk_registry,
        )

    @property
    def manager(self) -> PluginManager:
        """Access the underlying plugin manager.

        Returns:
            The PluginManager instance.
        """
        return self._manager

    async def discover(self) -> list[PluginManifest]:
        """Discover all available plugins.

        Returns:
            List of discovered plugin manifests.
        """
        return await self._manager.discover_plugins()

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-193"],
    )
    async def load(self, name: str) -> None:
        """Load a plugin by name.

        Args:
            name: Plugin name to load.
        """
        await self._manager.load_plugin(name)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-194"],
    )
    async def unload(self, name: str) -> None:
        """Unload a plugin by name.

        Args:
            name: Plugin name to unload.
        """
        await self._manager.unload_plugin(name)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-195"],
    )
    async def reload(self, name: str) -> None:
        """Hot-reload a plugin.

        Args:
            name: Plugin name to reload.
        """
        await self._manager.reload_plugin(name)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-192"],
    )
    def info(self, name: str) -> PluginInfo | None:
        """Get plugin info by name.

        Args:
            name: Plugin name.

        Returns:
            PluginInfo or None.
        """
        return self._manager.get_plugin(name)

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-191"],
    )
    def list_all(self) -> list[PluginInfo]:
        """List all known plugins.

        Returns:
            List of PluginInfo.
        """
        return self._manager.list_plugins()

    @traces(
        spec="015",
        story="US-014",
        reqs=["REQ-015-196"],
    )
    def active_plugins(self) -> list[PluginInfo]:
        """List only active plugins.

        Returns:
            List of PluginInfo with ACTIVE status.
        """
        return [
            p for p in self._manager.list_plugins() if p.status == PluginStatus.ACTIVE
        ]

    @traces(
        spec="015",
        story="US-007",
        reqs=["REQ-015-085"],
    )
    async def enable(self, name: str) -> None:
        """Enable a disabled plugin.

        Args:
            name: Plugin name.
        """
        await self._manager.enable_plugin(name)

    @traces(
        spec="015",
        story="US-007",
        reqs=["REQ-015-086"],
    )
    async def disable(self, name: str) -> None:
        """Disable an active plugin.

        Args:
            name: Plugin name.
        """
        await self._manager.disable_plugin(name)

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-002"],
    )
    async def install(self, source: str) -> PluginManifest:
        """Install a plugin from source.

        Args:
            source: Local path, pip spec, or git URL.

        Returns:
            The installed plugin's manifest.
        """
        return await self._manager.install_plugin(source)

    @traces(
        spec="015",
        story="US-007",
        reqs=["REQ-015-088"],
    )
    async def uninstall(self, name: str) -> None:
        """Uninstall a plugin completely.

        Args:
            name: Plugin name.
        """
        await self._manager.uninstall_plugin(name)

    @traces(
        spec="015",
        story="US-010",
        reqs=["REQ-015-129"],
    )
    async def start_all(self) -> None:
        """Start the plugin manager and all active plugins."""
        await self._manager.start()

    @traces(
        spec="015",
        story="US-010",
        reqs=["REQ-015-130"],
    )
    async def stop_all(self) -> None:
        """Stop the plugin manager and all active plugins."""
        await self._manager.stop()