metabuilder/workflow/plugins/python/registry.py

"""
Plugin registry for discovering and instantiating workflow plugins.

Plugins export NODE_TYPE, CATEGORY, DESCRIPTION, and impl - no side effects.
The registry handles lazy instantiation of executors when needed.
"""

import importlib
import pkgutil
from pathlib import Path
from typing import Any, Callable, Dict, Optional

from .base import NodeExecutor
from .factory import create_plugin


class PluginRegistry:
    """Registry for workflow plugins with lazy instantiation."""

    def __init__(self):
        self._plugins: Dict[str, dict] = {}  # node_type -> plugin metadata
        self._executors: Dict[str, NodeExecutor] = {}  # node_type -> executor (lazy)

    def register(
        self,
        node_type: str,
        category: str,
        description: str,
        impl: Callable[[Dict[str, Any], Any], Dict[str, Any]],
    ) -> None:
        """Register a plugin without instantiating it."""
        self._plugins[node_type] = {
            "node_type": node_type,
            "category": category,
            "description": description,
            "impl": impl,
        }

    def get_executor(self, node_type: str) -> Optional[NodeExecutor]:
        """Get or create an executor for a plugin (lazy instantiation)."""
        if node_type in self._executors:
            return self._executors[node_type]

        if node_type not in self._plugins:
            return None

        plugin = self._plugins[node_type]
        executor = create_plugin(
            plugin["node_type"],
            plugin["category"],
            plugin["description"],
            plugin["impl"],
        )
        self._executors[node_type] = executor
        return executor

    def run(
        self, node_type: str, runtime: Any, inputs: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Run a plugin by node type."""
        executor = self.get_executor(node_type)
        if executor is None:
            return {"error": f"Unknown plugin: {node_type}"}
        return executor.run(runtime, inputs)

    def list_plugins(self) -> Dict[str, dict]:
        """List all registered plugins."""
        return {
            node_type: {
                "node_type": p["node_type"],
                "category": p["category"],
                "description": p["description"],
            }
            for node_type, p in self._plugins.items()
        }

    def discover(self, package_path: str = None) -> None:
        """
        Discover and register all plugins from the plugins package.

        Scans for modules with NODE_TYPE, CATEGORY, DESCRIPTION, and impl.
        """
        if package_path is None:
            package_path = str(Path(__file__).parent)

        # Walk through all subdirectories
        for category_dir in Path(package_path).iterdir():
            if not category_dir.is_dir() or category_dir.name.startswith("_"):
                continue

            # Skip non-plugin directories
            if category_dir.name in ("__pycache__",):
                continue

            # Each subdirectory in category is a plugin
            for plugin_dir in category_dir.iterdir():
                if not plugin_dir.is_dir() or plugin_dir.name.startswith("_"):
                    continue

                # Look for the plugin module
                plugin_file = plugin_dir / f"{plugin_dir.name}.py"
                if not plugin_file.exists():
                    continue

                try:
                    # Import the module
                    module_name = f"workflow.plugins.python.{category_dir.name}.{plugin_dir.name}.{plugin_dir.name}"
                    module = importlib.import_module(module_name)

                    # Check for required exports
                    if hasattr(module, "NODE_TYPE") and hasattr(module, "impl"):
                        self.register(
                            node_type=module.NODE_TYPE,
                            category=getattr(module, "CATEGORY", category_dir.name),
                            description=getattr(module, "DESCRIPTION", ""),
                            impl=module.impl,
                        )
                except Exception:
                    # Skip plugins that fail to load
                    pass


# Global registry instance
registry = PluginRegistry()


def get_registry() -> PluginRegistry:
    """Get the global plugin registry."""
    return registry


def run_plugin(
    node_type: str, runtime: Any, inputs: Dict[str, Any]
) -> Dict[str, Any]:
    """Convenience function to run a plugin."""
    return registry.run(node_type, runtime, inputs)