File size: 3,577 Bytes
b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 | 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """Plugin registry - auto-discovery and registration of plugins."""
from __future__ import annotations
import importlib
import importlib.util
import sys
from pathlib import Path
from typing import TYPE_CHECKING
import yaml
from loguru import logger
if TYPE_CHECKING:
from plugins.base_plugin import BasePlugin
class PluginRegistry:
"""Registry for discovering and tracking plugin classes."""
def __init__(self):
self.plugin_classes: list[type[BasePlugin]] = []
self.metadata: dict[str, dict] = {}
def register_plugin_class(self, cls: type[BasePlugin]):
"""Register a plugin class (called automatically by __init_subclass__)."""
if cls not in self.plugin_classes:
self.plugin_classes.append(cls)
name = getattr(cls, "plugin_name", "") or cls.__name__
logger.debug(f"Registered plugin class: {name}")
def discover_and_load(self, plugins_dir: Path):
"""Discover and import plugins from a directory.
Each plugin is a subdirectory containing:
- metadata.yaml: Plugin metadata
- plugin.py: Plugin module with a BasePlugin subclass
"""
plugins_dir = Path(plugins_dir)
if not plugins_dir.exists():
logger.warning(f"Plugins directory does not exist: {plugins_dir}")
return
for plugin_dir in sorted(plugins_dir.iterdir()):
if not plugin_dir.is_dir():
continue
metadata_file = plugin_dir / "metadata.yaml"
plugin_file = plugin_dir / "plugin.py"
if not plugin_file.exists():
continue
# Load metadata
metadata = {}
if metadata_file.exists():
try:
with open(metadata_file, "r", encoding="utf-8") as f:
metadata = yaml.safe_load(f) or {}
except Exception as e:
logger.warning(f"Failed to load metadata for {plugin_dir.name}: {e}")
plugin_name = metadata.get("name", plugin_dir.name)
self.metadata[plugin_name] = metadata
# Import the plugin module
try:
self._import_plugin(plugin_dir.name, plugin_file)
logger.info(f"Loaded plugin: {plugin_name}")
except Exception as e:
logger.error(f"Failed to load plugin {plugin_name}: {e}")
def _import_plugin(self, name: str, plugin_file: Path):
"""Import a plugin module, triggering __init_subclass__ registration."""
module_name = f"plugins._loaded.{name}"
spec = importlib.util.spec_from_file_location(module_name, plugin_file)
if spec is None or spec.loader is None:
raise ImportError(f"Cannot create module spec for {plugin_file}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
def get_plugin_class(self, name: str) -> type[BasePlugin] | None:
"""Get a plugin class by name."""
for cls in self.plugin_classes:
cls_name = getattr(cls, "plugin_name", "") or cls.__name__
if cls_name == name:
return cls
return None
def list_plugins(self) -> list[str]:
"""List all registered plugin names."""
names = []
for cls in self.plugin_classes:
names.append(getattr(cls, "plugin_name", "") or cls.__name__)
return names
# Global registry instance
plugin_registry = PluginRegistry()
|