"""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 def init_app(self, app: Flask, db) -> None: """Initialize plugin manager with Flask app.""" self._app = app self._db = db # 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 _register_plugin_components(self, plugin: BasePlugin) -> None: """Register plugin's blueprint, models, CLI commands, etc.""" # Register blueprint blueprint = plugin.get_blueprint() if blueprint: self._app.register_blueprint( blueprint, url_prefix=plugin.meta.api_prefix ) logger.debug(f"Registered blueprint: {plugin.meta.api_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) 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 }) 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 # Load plugin class plugin_class = self.loader.load_plugin_class(name) if not plugin_class: logger.error(f"Plugin {name} not found") return False temp_plugin = plugin_class() meta = temp_plugin.meta # Check dependencies for dep in meta.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 self.registry.register(name, meta.version) # 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{meta.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 plugin_class = self.loader.load_plugin_class(name) if plugin_class: temp = plugin_class() for dep in temp.meta.dependencies: if not self.registry.is_enabled(dep): logger.error(f"Cannot enable {name}: {dep} is not enabled") return False self.registry.enable(name) # Load the plugin plugin = self.loader.load_plugin(name, self._app, self._db) if plugin: self._register_plugin_components(plugin) plugin.on_enable(self._app) 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() # Global plugin manager instance plugin_manager = PluginManager()