Spaces:
Sleeping
Sleeping
File size: 2,643 Bytes
5b6e956 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
"""
Plugin Manager - Discovers and loads all backend plugins.
"""
from pathlib import Path
from typing import Dict, List, Optional, Any
import importlib.util
import yaml
import sys
class PluginManager:
"""Manages all backend plugins."""
def __init__(self, plugin_dirs: List[Path]):
"""Initialize plugin manager."""
self.plugin_dirs = plugin_dirs
self.plugins: Dict[str, Any] = {}
self.discover_plugins()
def discover_plugins(self):
"""Automatically discover and load all plugins."""
for plugin_dir in self.plugin_dirs:
registry_path = plugin_dir / 'plugin_registry.yaml'
if not registry_path.exists():
continue
with open(registry_path) as f:
registry = yaml.safe_load(f)
for plugin_config in registry.get('plugins', []):
if plugin_config.get('enabled', True):
try:
self.load_plugin(plugin_dir, plugin_config)
except Exception as e:
print(f"Failed to load plugin {plugin_config['name']}: {e}")
def load_plugin(self, plugin_dir: Path, config: Dict):
"""Load a single plugin."""
plugin_name = config['name']
plugin_module = config['module']
plugin_class = config['class']
# Import plugin module
module_path = plugin_dir / f"{plugin_module}.py"
if not module_path.exists():
print(f"Plugin module not found: {module_path}")
return
spec = importlib.util.spec_from_file_location(plugin_module, module_path)
module = importlib.util.module_from_spec(spec)
sys.modules[plugin_module] = module
spec.loader.exec_module(module)
# Get plugin class
PluginClass = getattr(module, plugin_class)
# Find config path
config_path = plugin_dir.parent.parent / 'tools' / plugin_name / 'config.yaml'
# Instantiate plugin
plugin = PluginClass(config_path)
self.plugins[plugin_name] = plugin
print(f"Loaded plugin: {plugin_name} v{plugin.version}")
def get_plugin(self, name: str) -> Optional[Any]:
"""Get plugin by name."""
return self.plugins.get(name)
def list_plugins(self) -> List[str]:
"""List all available plugins."""
return list(self.plugins.keys())
def get_available_backends(self) -> Dict[str, bool]:
"""Get all backends with health status."""
return {
name: plugin.health_check()
for name, plugin in self.plugins.items()
}
|