shopdb-flask/shopdb/plugins/loader.py

"""Plugin discovery and loading."""

import importlib
import importlib.util
from pathlib import Path
from typing import Dict, List, Type, Optional
from flask import Flask
import logging

from .base import BasePlugin
from .registry import PluginRegistry

logger = logging.getLogger(__name__)


class PluginLoader:
    """
    Discovers and loads plugins from the plugins directory.
    """

    def __init__(self, plugins_dir: Path, registry: PluginRegistry):
        self.plugins_dir = plugins_dir
        self.registry = registry
        self._loaded_plugins: Dict[str, BasePlugin] = {}
        self._plugin_classes: Dict[str, Type[BasePlugin]] = {}

    def discover_plugins(self) -> List[str]:
        """
        Discover available plugins in plugins directory.
        Returns list of plugin names.
        """
        available = []
        if not self.plugins_dir.exists():
            return available

        for item in self.plugins_dir.iterdir():
            if item.is_dir() and (item / 'plugin.py').exists():
                available.append(item.name)
            elif item.is_dir() and (item / '__init__.py').exists():
                # Check for plugin.py in package
                if (item / 'plugin.py').exists():
                    available.append(item.name)

        return available

    def load_plugin_class(self, name: str) -> Optional[Type[BasePlugin]]:
        """
        Load plugin class without instantiating.
        Used for inspection before installation.
        """
        if name in self._plugin_classes:
            return self._plugin_classes[name]

        plugin_dir = self.plugins_dir / name
        plugin_module_path = plugin_dir / 'plugin.py'

        if not plugin_module_path.exists():
            logger.error(f"Plugin {name} not found: {plugin_module_path}")
            return None

        try:
            # Import the plugin module
            spec = importlib.util.spec_from_file_location(
                f"plugins.{name}.plugin",
                plugin_module_path
            )
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            # Find the plugin class
            for attr_name in dir(module):
                attr = getattr(module, attr_name)
                if (isinstance(attr, type) and
                    issubclass(attr, BasePlugin) and
                    attr is not BasePlugin):
                    self._plugin_classes[name] = attr
                    return attr

            logger.error(f"No BasePlugin subclass found in {name}")
            return None

        except Exception as e:
            logger.error(f"Error loading plugin {name}: {e}")
            return None

    def load_plugin(self, name: str, app: Flask, db) -> Optional[BasePlugin]:
        """
        Load and instantiate a plugin.
        """
        if name in self._loaded_plugins:
            return self._loaded_plugins[name]

        plugin_class = self.load_plugin_class(name)
        if not plugin_class:
            return None

        try:
            # Instantiate plugin
            plugin = plugin_class()

            # Check dependencies
            for dep in plugin.meta.dependencies:
                if not self.registry.is_enabled(dep):
                    logger.error(
                        f"Plugin {name} requires {dep} which is not enabled"
                    )
                    return None

            # Initialize plugin
            plugin.init_app(app, db)

            self._loaded_plugins[name] = plugin
            return plugin

        except Exception as e:
            logger.error(f"Error instantiating plugin {name}: {e}")
            return None

    def load_enabled_plugins(self, app: Flask, db) -> Dict[str, BasePlugin]:
        """
        Load all enabled plugins.
        Returns dict of name -> plugin instance.
        """
        loaded = {}

        # Sort by dependencies (simple topological sort)
        enabled = self.registry.get_enabled_plugins()
        sorted_plugins = self._sort_by_dependencies(enabled)

        for name in sorted_plugins:
            plugin = self.load_plugin(name, app, db)
            if plugin:
                loaded[name] = plugin
                logger.info(f"Loaded plugin: {name} v{plugin.meta.version}")
            else:
                logger.warning(f"Failed to load plugin: {name}")

        return loaded

    def _sort_by_dependencies(self, plugin_names: List[str]) -> List[str]:
        """Sort plugins so dependencies come first."""
        sorted_list = []
        visited = set()

        def visit(name):
            if name in visited:
                return
            visited.add(name)

            plugin_class = self.load_plugin_class(name)
            if plugin_class:
                # Create temporary instance to get meta
                try:
                    temp = plugin_class()
                    for dep in temp.meta.dependencies:
                        if dep in plugin_names:
                            visit(dep)
                except Exception:
                    pass

            sorted_list.append(name)

        for name in plugin_names:
            visit(name)

        return sorted_list

    def get_loaded_plugin(self, name: str) -> Optional[BasePlugin]:
        """Get an already loaded plugin."""
        return self._loaded_plugins.get(name)

    def get_all_loaded(self) -> Dict[str, BasePlugin]:
        """Get all loaded plugins."""
        return self._loaded_plugins.copy()