orchestrata_ultimate.plugin.registry

Reference › orchestrata_ultimate › plugin › registry

PluginRegistry is the thread-safe record of every plugin the PluginManager knows about. It keeps four maps keyed by plugin name: the plugin instances, their manifests, their PluginContext objects and their PluginInfo status records, all behind one lock.

register_discovered(manifest, source) adds a manifest and, if the name is new, an info record with status DISCOVERED. register(name, plugin, manifest, context, source) stores a loaded plugin with status ACTIVE and a load timestamp, and raises PluginRegistrationError if the name is already registered. unregister(name) calls the context's cleanup (which deregisters the plugin's Volks) before removing every entry, and raises PluginNotFoundError for an unknown name. set_status(name, status, error) updates the lifecycle status and error text. The remaining methods are lookups: get_plugin, get_info, get_manifest, get_context, list_plugins, contains and count. The status values are defined in interfaces.

At a glance

Class diagram

Dependency diagram

API

Classes

PluginRegistry

Line 25

In-memory registry of all known plugins.

Source

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

"""PluginRegistry — tracks loaded plugins and their metadata.

Provides thread-safe storage for plugin instances, manifests,
contexts, and runtime info. The single source of truth for
which plugins are loaded and their current status.
"""

import threading
from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.plugin.context import PluginContext
from orchestrata_ultimate.plugin.exceptions import (
    PluginNotFoundError,
    PluginRegistrationError,
)
from orchestrata_ultimate.plugin.interfaces import (
    IPlugin,
    PluginInfo,
    PluginStatus,
)
from orchestrata_ultimate.plugin.manifest import PluginManifest


class PluginRegistry:
    """In-memory registry of all known plugins.

    Thread-safe storage for plugin instances, manifests, contexts,
    and runtime state. Supports register, unregister, list, and
    lookup operations.
    """

    def __init__(self) -> None:
        """Initialize an empty plugin registry."""
        self._lock = threading.Lock()
        self._plugins: dict[str, IPlugin] = {}
        self._manifests: dict[str, PluginManifest] = {}
        self._contexts: dict[str, PluginContext] = {}
        self._infos: dict[str, PluginInfo] = {}

    @traces(spec="015", story="US-014", reqs=["REQ-015-193"])
    def register(
        self,
        name: str,
        plugin: IPlugin,
        manifest: PluginManifest,
        context: PluginContext,
        source: str = "",
    ) -> None:
        """Register a loaded plugin.

        Args:
            name: Unique plugin name.
            plugin: Plugin instance implementing IPlugin.
            manifest: Validated plugin manifest.
            context: Plugin context for DI and registration.
            source: Where the plugin was found.

        Raises:
            PluginRegistrationError: If name is already registered.
        """
        with self._lock:
            if name in self._plugins:
                raise PluginRegistrationError(
                    f"Plugin '{name}' is already registered",
                    context={"name": name},
                )
            self._plugins[name] = plugin
            self._manifests[name] = manifest
            self._contexts[name] = context
            self._infos[name] = PluginInfo(
                name=name,
                version=manifest.version,
                description=manifest.description,
                author=manifest.author,
                status=PluginStatus.ACTIVE,
                source=source,
                loaded_at=datetime.now(UTC),
            )

    @traces(
        spec="015",
        story="US-001",
        reqs=["REQ-015-009"],
    )
    def register_discovered(
        self,
        manifest: PluginManifest,
        source: str = "",
    ) -> None:
        """Register a discovered but not yet loaded plugin.

        Args:
            manifest: Validated plugin manifest.
            source: Where the plugin was found.
        """
        with self._lock:
            name = manifest.name
            if name not in self._infos:
                self._infos[name] = PluginInfo(
                    name=name,
                    version=manifest.version,
                    description=manifest.description,
                    author=manifest.author,
                    status=PluginStatus.DISCOVERED,
                    source=source,
                )
            self._manifests[name] = manifest

    @traces(spec="015", story="US-010", reqs=["REQ-015-127", "REQ-015-088"])
    def unregister(self, name: str) -> None:
        """Remove a plugin from the registry.

        Cleans up the plugin context (deregisters Volks, tools,
        etc.) before removal.

        Args:
            name: Plugin name to remove.

        Raises:
            PluginNotFoundError: If plugin is not registered.
        """
        with self._lock:
            if name not in self._infos:
                raise PluginNotFoundError(
                    f"Plugin '{name}' not found in registry",
                    context={"name": name},
                )
            ctx = self._contexts.pop(name, None)
            if ctx:
                ctx.cleanup()
            self._plugins.pop(name, None)
            self._manifests.pop(name, None)
            self._infos.pop(name, None)

    def get_plugin(self, name: str) -> IPlugin | None:
        """Get a plugin instance by name.

        Args:
            name: Plugin name.

        Returns:
            Plugin instance or None if not loaded.
        """
        with self._lock:
            return self._plugins.get(name)

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

        Args:
            name: Plugin name.

        Returns:
            PluginInfo or None if not known.
        """
        with self._lock:
            return self._infos.get(name)

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

        Args:
            name: Plugin name.

        Returns:
            PluginManifest or None if not known.
        """
        with self._lock:
            return self._manifests.get(name)

    def get_context(self, name: str) -> PluginContext | None:
        """Get plugin context by name.

        Args:
            name: Plugin name.

        Returns:
            PluginContext or None if not loaded.
        """
        with self._lock:
            return self._contexts.get(name)

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

        Returns:
            List of PluginInfo for all discovered/loaded plugins.
        """
        with self._lock:
            return list(self._infos.values())

    @traces(spec="015", story="US-007", reqs=["REQ-015-094"])
    def set_status(
        self,
        name: str,
        status: PluginStatus,
        error: str | None = None,
    ) -> None:
        """Update the status of a plugin.

        Args:
            name: Plugin name.
            status: New status to set.
            error: Optional error message for ERROR status.

        Raises:
            PluginNotFoundError: If plugin is not known.
        """
        with self._lock:
            info = self._infos.get(name)
            if info is None:
                raise PluginNotFoundError(
                    f"Plugin '{name}' not found in registry",
                    context={"name": name},
                )
            info.status = status
            info.error = error

    def contains(self, name: str) -> bool:
        """Check if a plugin is known to the registry.

        Args:
            name: Plugin name to check.

        Returns:
            True if the plugin is known (discovered or loaded).
        """
        with self._lock:
            return name in self._infos

    def count(self) -> int:
        """Return the number of known plugins.

        Returns:
            Number of plugins in the registry.
        """
        with self._lock:
            return len(self._infos)