"""Plugin manager - main entry point for plugin system.""" from pathlib import Path from typing import Dict, List, Optional from flask import Flask import logging from .base import BasePlugin, PluginMeta from .registry import PluginRegistry, PluginState from .loader import PluginLoader from .migrations import PluginMigrationManager logger = logging.getLogger(__name__) __all__ = [ 'PluginManager', 'BasePlugin', 'PluginMeta', 'PluginRegistry', 'PluginState', 'plugin_manager' ] class PluginManager: """ Central manager for all plugin operations. Usage: plugin_manager = PluginManager() plugin_manager.init_app(app, db) # In CLI: plugin_manager.install_plugin('printers') """ def __init__(self): self.registry: Optional[PluginRegistry] = None self.loader: Optional[PluginLoader] = None self.migration_manager: Optional[PluginMigrationManager] = None self._app: Optional[Flask] = None self._db = None # API prefixes already claimed by a registered plugin blueprint, to # detect two plugins overlapping on the same /api/... namespace. self._registered_prefixes: set = set() def init_app(self, app: Flask, db) -> None: """Initialize plugin manager with Flask app.""" self._app = app self._db = db # Reset per-app so the prefix-uniqueness guard tracks only this app's # registrations (the manager is a process-wide singleton; tests build # multiple apps from it). self._registered_prefixes = set() # Setup paths instance_path = Path(app.instance_path) plugins_dir = Path(app.root_path).parent / 'plugins' # Initialize components self.registry = PluginRegistry(instance_path / 'plugins.json') self.loader = PluginLoader(plugins_dir, self.registry) self.migration_manager = PluginMigrationManager( plugins_dir, app.config.get('SQLALCHEMY_DATABASE_URI') ) # Load enabled plugins self._load_enabled_plugins() # Store on app for access app.extensions['plugin_manager'] = self def _load_enabled_plugins(self) -> None: """Load and register all enabled plugins.""" plugins = self.loader.load_enabled_plugins(self._app, self._db) for name, plugin in plugins.items(): self._register_plugin_components(plugin) def upgrade_all_plugins(self) -> Dict[str, str]: """Run pending Alembic migrations for every loaded plugin. Returns {plugin_name: 'ok'|'no-migrations'|}. Skips plugins with no migrations/ directory. Use from the CLI (`flask plugin upgrade-all`) on a fresh deploy after the core schema is in place; existing deploys that still use db.create_all can ignore this and continue to do so. """ results: Dict[str, str] = {} if not self.migration_manager: return results plugin_names = list(self.registry.list_installed().keys()) \ if hasattr(self.registry, 'list_installed') else [] if not plugin_names: # Fall back to whatever the loader found on disk. plugin_names = self.loader.discover_plugins() for name in plugin_names: if not self.migration_manager.has_pending_migrations(name): results[name] = 'no-migrations' continue try: ok = self.migration_manager.run_plugin_migrations(name) results[name] = 'ok' if ok else 'failed' except Exception as ex: results[name] = f'error: {ex}' return results def _register_plugin_components(self, plugin: BasePlugin) -> None: """Register plugin's blueprint, models, CLI commands, etc.""" # Register blueprint blueprint = plugin.get_blueprint() if blueprint: prefix = plugin.meta.api_prefix # Guard against two plugins claiming the same API prefix; Flask only # rejects duplicate blueprint names, not overlapping url_prefixes, so # an overlap would silently shadow routes. if prefix in self._registered_prefixes: raise ValueError( f"Plugin {plugin.meta.name} api_prefix '{prefix}' is already " f"claimed by another blueprint" ) self._app.register_blueprint(blueprint, url_prefix=prefix) self._registered_prefixes.add(prefix) logger.debug(f"Registered blueprint: {prefix}") # Register CLI commands for cmd in plugin.get_cli_commands(): self._app.cli.add_command(cmd) def discover_available(self) -> List[Dict]: """ Get list of all available plugins (installed or not). Returns list of plugin info dicts. """ available = [] for name in self.loader.discover_plugins(): plugin_class = self.loader.load_plugin_class(name) if plugin_class: try: temp = plugin_class() meta = temp.meta state = self.registry.get(name) try: config_schema = temp.get_config_schema() except Exception: config_schema = [] try: provisioning_note = temp.get_provisioning_note() except Exception: provisioning_note = None manifest = self.loader.load_manifest(name) available.append({ 'name': meta.name, 'version': meta.version, 'description': meta.description, 'author': meta.author, 'dependencies': meta.dependencies, 'installed': state is not None, 'enabled': state.enabled if state else False, 'installedat': state.installed_at if state else None, 'config_schema': config_schema, 'provisioning_note': provisioning_note, 'default_enabled': manifest.get('default_enabled', True), }) except Exception as e: logger.warning(f"Error inspecting plugin {name}: {e}") return available def install_plugin(self, name: str, run_migrations: bool = True) -> bool: """ Install a plugin. Steps: 1. Verify plugin exists 2. Check dependencies 3. Run database migrations 4. Register in registry 5. Call plugin's on_install hook """ # Check if already installed if self.registry.is_installed(name): logger.warning(f"Plugin {name} is already installed") return False # Read metadata from the manifest (single source of truth) instead of # instantiating the plugin class just to inspect deps/version. manifest = self.loader.load_manifest(name) if not manifest: logger.error(f"Plugin {name} not found") return False manifest_version = manifest.get('version') # Check dependencies for dep in manifest.get('dependencies', []): if not self.registry.is_installed(dep): logger.error( f"Plugin {name} requires {dep} to be installed first" ) return False # Run migrations if run_migrations: success = self.migration_manager.run_plugin_migrations(name) if not success: logger.error(f"Failed to run migrations for {name}") return False # Register plugin # Plugins that provision extra tables install disabled until a site # opts in (manifest default_enabled=false). self.registry.register(name, manifest_version, enabled=manifest.get('default_enabled', True)) # Load the plugin plugin = self.loader.load_plugin(name, self._app, self._db) if plugin: self._register_plugin_components(plugin) plugin.on_install(self._app) logger.info(f"Installed plugin: {name} v{manifest_version}") return True def uninstall_plugin(self, name: str, remove_data: bool = False) -> bool: """ Uninstall a plugin. Args: name: Plugin name remove_data: If True, run downgrade migrations to remove tables """ if not self.registry.is_installed(name): logger.warning(f"Plugin {name} is not installed") return False # Check if other plugins depend on this one for other_name in self.registry.get_enabled_plugins(): if other_name == name: continue other_plugin = self.loader.get_loaded_plugin(other_name) if other_plugin and name in other_plugin.meta.dependencies: logger.error( f"Cannot uninstall {name}: {other_name} depends on it" ) return False # Get plugin instance plugin = self.loader.get_loaded_plugin(name) # Call on_uninstall hook if plugin: plugin.on_uninstall(self._app) # Optionally remove data if remove_data: self.migration_manager.downgrade_plugin(name) # Unregister self.registry.unregister(name) logger.info(f"Uninstalled plugin: {name}") return True def enable_plugin(self, name: str) -> bool: """Enable a disabled plugin.""" if not self.registry.is_installed(name): logger.error(f"Plugin {name} is not installed") return False if self.registry.is_enabled(name): logger.info(f"Plugin {name} is already enabled") return True # Check dependencies are enabled. Read deps from the manifest, not by # instantiating the plugin class (manifest is the single source of # truth; instantiating fires __init__ side effects unnecessarily). manifest = self.loader.load_manifest(name) for dep in manifest.get('dependencies', []): if not self.registry.is_enabled(dep): logger.error(f"Cannot enable {name}: {dep} is not enabled") return False self.registry.enable(name) # Fire the on_enable hook best-effort. Do NOT register the blueprint # here: Flask forbids register_blueprint after the first request, so # routes/nav for a re-enabled plugin take effect on the next restart # (symmetric with disable). try: plugin = self.loader.load_plugin(name, self._app, self._db) if plugin: plugin.on_enable(self._app) except Exception: logger.exception(f"on_enable hook failed for plugin {name}") logger.info(f"Enabled plugin: {name}") return True def disable_plugin(self, name: str) -> bool: """Disable an enabled plugin.""" if not self.registry.is_enabled(name): logger.info(f"Plugin {name} is already disabled") return True # Check if other plugins depend on this one for other_name in self.registry.get_enabled_plugins(): if other_name == name: continue other_plugin = self.loader.get_loaded_plugin(other_name) if other_plugin and name in other_plugin.meta.dependencies: logger.error( f"Cannot disable {name}: {other_name} depends on it" ) return False plugin = self.loader.get_loaded_plugin(name) if plugin: plugin.on_disable(self._app) self.registry.disable(name) logger.info(f"Disabled plugin: {name}") return True def get_plugin(self, name: str) -> Optional[BasePlugin]: """Get a loaded plugin instance.""" return self.loader.get_loaded_plugin(name) def get_all_plugins(self) -> Dict[str, BasePlugin]: """Get all loaded plugins.""" return self.loader.get_all_loaded() def get_service(self, name: str): """Resolve a service exposed by an enabled plugin via get_services(). Consumer for the BasePlugin.get_services hook: searches enabled plugins for one that registers `name` and returns the registered value (a service class or factory). Returns None if no enabled plugin provides it. This is how one plugin obtains another's service (e.g. the Zabbix service). """ for plugin_name, plugin in self.get_all_plugins().items(): if not self.registry.is_enabled(plugin_name): continue try: services = plugin.get_services() or {} except Exception: continue if name in services: return services[name] return None # Global plugin manager instance plugin_manager = PluginManager()