"""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()