""" Plugin registry for discovering and managing backend plugins. This module handles filesystem-based plugin discovery, validation, and role-based filtering. """ import importlib.util import logging from pathlib import Path from typing import Any from fastapi_app.lib.plugins.plugin_base import Plugin, PluginContext logger = logging.getLogger(__name__) class PluginRegistry: """ Registry for discovering and managing plugins from the filesystem. """ def __init__(self): """Initialize empty plugin registry.""" self._plugins: dict[str, Plugin] = {} self._plugin_metadata: dict[str, dict[str, Any]] = {} def discover_plugins(self, plugin_dirs: list[Path]) -> None: """ Discover plugins from the specified directories. Each plugin directory should contain a plugin.py file with a Plugin subclass. Uses two-phase registration to handle dependencies: 1. Load all plugins 2. Register in dependency order Args: plugin_dirs: List of directories to search for plugins """ pending_plugins: list[Plugin] = [] for plugin_dir in plugin_dirs: if not plugin_dir.exists(): logger.warning(f"Plugin directory does not exist: {plugin_dir}") continue if not plugin_dir.is_dir(): logger.warning(f"Plugin path is not a directory: {plugin_dir}") continue # Iterate through subdirectories (each is a potential plugin) for plugin_path in plugin_dir.iterdir(): if not plugin_path.is_dir(): continue plugin_file = plugin_path / "plugin.py" if not plugin_file.exists(): continue try: plugin = self._load_plugin(plugin_path, plugin_file) if plugin: pending_plugins.append(plugin) except Exception as e: logger.error(f"Failed to load plugin from {plugin_path}: {e}") # Second pass: register with dependency resolution self._register_with_dependencies(pending_plugins) def _load_plugin(self, plugin_path: Path, plugin_file: Path) -> Plugin | None: """ Load a plugin from a plugin.py file. Args: plugin_path: Path to plugin directory plugin_file: Path to plugin.py file Returns: Plugin instance or None if loading failed """ try: # Use standard package import to support relative imports plugin_name = plugin_path.name module_name = f"fastapi_app.plugins.{plugin_name}.plugin" # Try standard import first (works when fastapi_app is in sys.path) try: import importlib module = importlib.import_module(module_name) except ModuleNotFoundError: # Fallback to spec-based loading for edge cases spec = importlib.util.spec_from_file_location(module_name, plugin_file) if spec is None or spec.loader is None: logger.error(f"Could not load spec for {plugin_file}") return None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # Find Plugin subclass in module plugin_class = None for item_name in dir(module): item = getattr(module, item_name) if ( isinstance(item, type) and issubclass(item, Plugin) and item is not Plugin ): plugin_class = item break if plugin_class is None: logger.warning(f"No Plugin subclass found in {plugin_file}") return None # Instantiate first so __init__ can populate config (e.g. via get_plugin_config), # then check availability which may depend on that config being written. plugin_instance = plugin_class() if not plugin_class.is_available(): logger.info(f"Plugin in {plugin_file} is not available, skipping") return None return plugin_instance except Exception as e: logger.error(f"Error loading plugin from {plugin_file}: {e}") return None def _register_plugin(self, plugin: Plugin) -> None: """ Register a plugin instance. Args: plugin: Plugin instance to register Raises: ValueError: If plugin metadata is invalid """ # Validate metadata metadata = plugin.metadata required_fields = ["id", "name", "description", "category", "version", "required_roles"] for field in required_fields: if field not in metadata: raise ValueError(f"Plugin metadata missing required field: {field}") plugin_id = metadata["id"] # Check for duplicate IDs if plugin_id in self._plugins: logger.warning(f"Plugin with id '{plugin_id}' already registered, skipping") return # Register self._plugins[plugin_id] = plugin self._plugin_metadata[plugin_id] = metadata logger.info(f"Registered plugin: {plugin_id} ({metadata['name']})") def _register_with_dependencies(self, plugins: list[Plugin]) -> None: """ Register plugins in dependency order using topological sort. Args: plugins: List of plugin instances to register """ # Build plugin map by ID plugin_map = {p.metadata["id"]: p for p in plugins} registered: set[str] = set() def register_plugin(plugin_id: str, path: list[str]) -> bool: """ Recursively register a plugin and its dependencies. Args: plugin_id: Plugin ID to register path: Current dependency path (for cycle detection) Returns: True if registration succeeded, False otherwise """ if plugin_id in registered: return True if plugin_id in path: cycle = " -> ".join(path + [plugin_id]) logger.error(f"Circular dependency detected: {cycle}") return False if plugin_id not in plugin_map: logger.error(f"Missing dependency: {plugin_id}") return False plugin = plugin_map[plugin_id] deps = plugin.metadata.get("dependencies", []) # Register dependencies first for dep_id in deps: if not register_plugin(dep_id, path + [plugin_id]): logger.error( f"Plugin {plugin_id} not registered due to " f"failed dependency: {dep_id}" ) return False # Now register this plugin try: self._register_plugin(plugin) registered.add(plugin_id) return True except Exception as e: logger.error(f"Failed to register plugin {plugin_id}: {e}") return False # Register all plugins for plugin_id in plugin_map: register_plugin(plugin_id, []) def get_dependency(self, plugin_id: str, dependency_id: str) -> Plugin | None: """ Get a dependency plugin instance. Only returns the dependency if it was declared in the requesting plugin's metadata. Logs a warning if an undeclared dependency is requested. Args: plugin_id: ID of the requesting plugin dependency_id: ID of the dependency to retrieve Returns: Plugin instance or None if not a declared dependency """ plugin = self._plugins.get(plugin_id) if not plugin: return None deps = plugin.metadata.get("dependencies", []) if dependency_id not in deps: logger.warning( f"Plugin {plugin_id} requested undeclared dependency {dependency_id}" ) return None return self._plugins.get(dependency_id) def get_plugin(self, plugin_id: str) -> Plugin | None: """ Get a plugin by ID. Args: plugin_id: Plugin identifier Returns: Plugin instance or None if not found """ return self._plugins.get(plugin_id) def get_plugins( self, category: str | None = None, user_roles: list[str] | None = None ) -> list[dict[str, Any]]: """ Get plugin metadata filtered by category and user roles. Args: category: Optional category filter user_roles: User's roles for access control (None = no filtering) Returns: List of plugin metadata dicts """ filtered = [] for plugin_id, metadata in self._plugin_metadata.items(): # Filter by category if category and metadata.get("category") != category: continue # Filter by required roles (only if user_roles is explicitly provided) if user_roles is not None: required_roles = metadata.get("required_roles", []) # If plugin requires specific roles (not empty and not just "*") if required_roles and required_roles != ["*"]: # Check if user has wildcard role or any of the required roles has_access = '*' in user_roles or any(role in user_roles for role in required_roles) if not has_access: continue # If plugin has required_roles=[] or ["*"], it's accessible to everyone # Include in results (exclude internal fields if needed) result = { "id": metadata["id"], "name": metadata["name"], "description": metadata["description"], "category": metadata["category"], "version": metadata["version"], } # Include optional endpoints metadata if present if "endpoints" in metadata: result["endpoints"] = metadata["endpoints"] filtered.append(result) return filtered def get_all_plugins(self) -> dict[str, Plugin]: """ Get all registered plugins (no filtering). Returns: Dict mapping plugin IDs to Plugin instances """ return self._plugins.copy() async def initialize_all(self, context: Any) -> None: """ Initialize all registered plugins. Creates a per-plugin context with plugin_id and registry so that plugins can access their declared dependencies during initialization. Args: context: Base plugin context (provides app reference) """ for plugin_id, plugin in self._plugins.items(): try: plugin_context = PluginContext( app=context.app, plugin_id=plugin_id, registry=self, ) await plugin.initialize(plugin_context) metadata = self._plugin_metadata.get(plugin_id, {}) logger.info(f"Initialized plugin: {plugin_id} ({metadata.get('name', 'unknown')})") except Exception as e: logger.error(f"Error initializing plugin {plugin_id}: {e}") async def cleanup_all(self) -> None: """ Cleanup all registered plugins. """ for plugin_id, plugin in self._plugins.items(): try: await plugin.cleanup() except Exception as e: logger.error(f"Error cleaning up plugin {plugin_id}: {e}")