""" Plugin manager for application-wide plugin lifecycle management. This module provides a singleton manager that handles plugin discovery, initialization, route registration, and execution. """ import logging import os from pathlib import Path from typing import Any from fastapi import FastAPI from fastapi_app.config import get_settings from fastapi_app.lib.plugins.plugin_base import PluginContext from fastapi_app.lib.plugins.plugin_registry import PluginRegistry logger = logging.getLogger(__name__) class PluginManager: """ Singleton manager for plugin lifecycle and route registration. """ _instance: "PluginManager | None" = None def __init__(self): """Initialize plugin manager (use get_instance() instead).""" if PluginManager._instance is not None: raise RuntimeError("Use PluginManager.get_instance() instead") self.registry = PluginRegistry() self._app: FastAPI | None = None self._initialized = False @classmethod def get_instance(cls) -> "PluginManager": """ Get or create the singleton plugin manager instance. Returns: PluginManager singleton instance """ if cls._instance is None: cls._instance = cls() return cls._instance def discover_plugins(self) -> None: """ Discover plugins from configured directories. Searches: 1. Built-in plugins: fastapi_app/plugins/ 2. Additional paths from FASTAPI_PLUGIN_PATHS environment variable """ plugin_dirs: list[Path] = [] # Built-in plugins directory builtin_dir = get_settings().plugins_code_dir plugin_dirs.append(builtin_dir) # Additional plugin paths from environment env_paths = os.getenv("FASTAPI_PLUGIN_PATHS", "") if env_paths: # Split by colon on Unix, semicolon on Windows separator = ";" if os.name == "nt" else ":" for path_str in env_paths.split(separator): path_str = path_str.strip() if path_str: plugin_dirs.append(Path(path_str)) logger.info(f"Discovering plugins from {len(plugin_dirs)} directories") self.registry.discover_plugins(plugin_dirs) def register_plugin_routes(self, app: FastAPI) -> None: """ Register custom routes from plugins that have a routes.py file. Also mounts static file directories for plugins with a 'static' subdirectory. Args: app: FastAPI application instance """ self._app = app # Iterate through all registered plugins and try to register their routes for plugin_id in self.registry.get_all_plugins().keys(): try: self._try_register_plugin_routes(app, plugin_id) self._try_mount_plugin_static_files(app, plugin_id) except Exception as e: logger.error(f"Error registering routes for plugin {plugin_id}: {e}") def _try_register_plugin_routes(self, app: FastAPI, plugin_id: str) -> None: """ Try to load and register routes.py for a plugin. Args: app: FastAPI application instance plugin_id: Plugin identifier """ # Search plugin directories for this plugin's routes.py plugin_dirs = self._get_plugin_dirs() # Try both hyphenated and underscored directory names # (plugin_id uses hyphens, directory names use underscores) dir_names = [plugin_id, plugin_id.replace("-", "_")] for base_dir in plugin_dirs: for dir_name in dir_names: plugin_dir = base_dir / dir_name routes_file = plugin_dir / "routes.py" if routes_file.exists(): try: # Load the routes module using its proper dotted module name # so relative imports work correctly. import importlib import importlib.util import sys project_root = get_settings().project_root_dir rel = routes_file.relative_to(project_root).with_suffix("") module_name = ".".join(rel.parts) package_name = ".".join(rel.parts[:-1]) # Ensure all parent packages are importable if package_name not in sys.modules: importlib.import_module(package_name) spec = importlib.util.spec_from_file_location( module_name, routes_file, submodule_search_locations=[] ) if spec is None or spec.loader is None: logger.warning(f"Could not create spec for {routes_file}") continue module = importlib.util.module_from_spec(spec) module.__package__ = package_name sys.modules[module_name] = module spec.loader.exec_module(module) # Look for 'router' in the module if hasattr(module, "router"): app.include_router(module.router) logger.info(f"Registered custom routes for plugin: {plugin_id}") return else: logger.warning(f"No 'router' found in {routes_file}") except Exception as e: logger.error( f"Error loading routes.py for plugin {plugin_id}: {e}", exc_info=True ) def _try_mount_plugin_static_files(self, app: FastAPI, plugin_id: str) -> None: """ Try to mount static files from plugin's static directory. Args: app: FastAPI application instance plugin_id: Plugin identifier """ from fastapi.staticfiles import StaticFiles plugin_dirs = self._get_plugin_dirs() dir_names = [plugin_id, plugin_id.replace("-", "_")] for base_dir in plugin_dirs: for dir_name in dir_names: plugin_dir = base_dir / dir_name static_dir = plugin_dir / "static" if static_dir.exists() and static_dir.is_dir(): try: # Mount at /api/plugins/{plugin_id}/static/ mount_path = f"/api/plugins/{plugin_id}/static" app.mount( mount_path, StaticFiles(directory=str(static_dir)), name=f"plugin_{plugin_id}_static" ) logger.info(f"Mounted static files for plugin {plugin_id} at {mount_path}") return except Exception as e: logger.error(f"Error mounting static files for plugin {plugin_id}: {e}") def _get_plugin_dirs(self) -> list[Path]: """ Get list of plugin directories to search. Returns: List of plugin directory paths """ plugin_dirs: list[Path] = [] # Built-in plugins builtin_dir = get_settings().plugins_code_dir plugin_dirs.append(builtin_dir) # Additional paths from environment env_paths = os.getenv("FASTAPI_PLUGIN_PATHS", "") if env_paths: separator = ";" if os.name == "nt" else ":" for path_str in env_paths.split(separator): path_str = path_str.strip() if path_str: plugin_dirs.append(Path(path_str)) return plugin_dirs async def initialize_plugins(self, app: FastAPI) -> None: """ Initialize all plugins with application context. Args: app: FastAPI application instance """ if self._initialized: return context = PluginContext(app=app) await self.registry.initialize_all(context) self._initialized = True logger.info("All plugins initialized") async def shutdown_plugins(self) -> None: """ Cleanup all plugins on application shutdown. """ await self.registry.cleanup_all() logger.info("All plugins cleaned up") 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 Returns: List of plugin metadata dicts """ return self.registry.get_plugins(category=category, user_roles=user_roles) async def execute_plugin( self, plugin_id: str, endpoint: str, params: dict[str, Any], user: dict | None = None ) -> Any: """ Execute a plugin endpoint. Args: plugin_id: Plugin identifier endpoint: Endpoint name to execute params: Parameters to pass to endpoint user: Current user dict (for context) Returns: Result from plugin endpoint Raises: ValueError: If plugin or endpoint not found """ plugin = self.registry.get_plugin(plugin_id) if plugin is None: raise ValueError(f"Plugin not found: {plugin_id}") endpoints = plugin.get_endpoints() if endpoint not in endpoints: raise ValueError(f"Endpoint not found: {plugin_id}.{endpoint}") endpoint_func = endpoints[endpoint] # Create context for this execution with plugin_id and registry for dependency access context = PluginContext( app=self._app, user=user, plugin_id=plugin_id, registry=self.registry, ) # Execute endpoint (pass both context and params) return await endpoint_func(context, params)