orchestrata_ultimate.plugin.config

Reference › orchestrata_ultimate › plugin › config

The configuration of the plugin system. PluginConfig holds: enabled (default true), plugin_dirs (default plugins), auto_discover (default true), allowlist_mode (open by default, or explicit), the allowlist and blocklist of plugin names, the resource_limits, an error_threshold of 5 consecutive errors, per-plugin configuration dictionaries keyed by plugin name, and the module prefixes plugin entry points may import from. PluginResourceLimits carries the defaults per plugin: 512 MiB of memory, 10,000 API calls per hour and 5,000 events per hour. PluginConfig.from_dict builds the object from a plain dictionary, filling missing keys with these defaults. The PluginManager takes it, the loader uses the directories and prefixes, and governance uses the lists and limits.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PluginResourceLimits (dataclass)

Decorators: traces(spec='015', story='US-009', reqs=['REQ-015-119']), dataclass · Line 20

Resource limits applied to each plugin.

PluginConfig (dataclass)

Decorators: traces(spec='015', story='US-009', reqs=['REQ-015-113']), dataclass · Line 36

Configuration for the PluginManager.

Source

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

"""Plugin configuration dataclass.

Defines the PluginConfig that controls PluginManager behavior:
discovery directories, allowlist/blocklist, resource limits,
allowed entry point prefixes, and error thresholds.
"""

from dataclasses import dataclass, field

from orchestrata_ultimate.core.base.traces import traces

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


@traces(spec="015", story="US-009", reqs=["REQ-015-119"])
@dataclass
class PluginResourceLimits:
    """Resource limits applied to each plugin.

    Attributes:
        default_memory_bytes: Max memory per plugin (default 512MB).
        default_api_calls_per_hour: Max API calls per hour.
        default_events_per_hour: Max events emitted per hour.
    """

    default_memory_bytes: int = 536_870_912
    default_api_calls_per_hour: int = 10_000
    default_events_per_hour: int = 5_000


@traces(spec="015", story="US-009", reqs=["REQ-015-113"])
@dataclass
class PluginConfig:
    """Configuration for the PluginManager.

    Attributes:
        enabled: Whether the plugin system is active.
        plugin_dirs: Directories to scan for local plugins.
        auto_discover: Whether to auto-discover on startup.
        allowlist_mode: "explicit" requires allowlist entry;
            "open" allows any non-blocklisted plugin.
        allowlist: Plugin names allowed to load.
        blocklist: Plugin names blocked from loading.
        resource_limits: Default resource limits per plugin.
        error_threshold: Consecutive errors before auto-disable.
        plugin_configs: Per-plugin configuration dicts keyed by
            plugin name.
        allowed_prefixes: Module path prefixes allowed for
            plugin entry points.
    """

    enabled: bool = True
    plugin_dirs: list[str] = field(default_factory=lambda: ["plugins"])
    auto_discover: bool = True
    allowlist_mode: str = "open"
    allowlist: list[str] = field(default_factory=list)
    blocklist: list[str] = field(default_factory=list)
    resource_limits: PluginResourceLimits = field(default_factory=PluginResourceLimits)
    error_threshold: int = 5
    plugin_configs: dict[str, dict] = field(default_factory=dict)
    allowed_prefixes: tuple[str, ...] = DEFAULT_ALLOWED_PREFIXES

    @staticmethod
    @traces(spec="015", story="US-008", reqs=["REQ-015-096"])
    def from_dict(data: dict) -> PluginConfig:
        """Create PluginConfig from a configuration dictionary.

        Args:
            data: Dictionary with plugin manager configuration.

        Returns:
            Populated PluginConfig instance.
        """
        limits_data = data.get("resource_limits", {})
        limits = PluginResourceLimits(
            default_memory_bytes=limits_data.get("default_memory_bytes", 536_870_912),
            default_api_calls_per_hour=limits_data.get(
                "default_api_calls_per_hour", 10_000
            ),
            default_events_per_hour=limits_data.get("default_events_per_hour", 5_000),
        )
        prefixes_raw = data.get("allowed_prefixes")
        prefixes = (
            tuple(prefixes_raw)
            if prefixes_raw is not None
            else DEFAULT_ALLOWED_PREFIXES
        )
        return PluginConfig(
            enabled=data.get("enabled", True),
            plugin_dirs=data.get("plugin_dirs", ["plugins"]),
            auto_discover=data.get("auto_discover", True),
            allowlist_mode=data.get("allowlist_mode", "open"),
            allowlist=data.get("allowlist", []),
            blocklist=data.get("blocklist", []),
            resource_limits=limits,
            error_threshold=data.get("error_threshold", 5),
            plugin_configs=data.get("plugin_configs", {}),
            allowed_prefixes=prefixes,
        )