72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
from typing import List, Optional
|
|
|
|
from .plugin import Plugin, PluginRegistry
|
|
|
|
|
|
class PluginLoader:
|
|
def __init__(self, registry: Optional[PluginRegistry] = None):
|
|
self.registry = registry or PluginRegistry()
|
|
self._entry_point_name = 'errorfix.plugins'
|
|
|
|
def load_entry_points(self) -> List[Plugin]:
|
|
plugins = []
|
|
try:
|
|
import importlib.metadata
|
|
eps = importlib.metadata.entry_points()
|
|
if hasattr(eps, 'select'):
|
|
entry_points = eps.select(group=self._entry_point_name)
|
|
else:
|
|
entry_points = eps.get(self._entry_point_name, [])
|
|
|
|
for ep in entry_points:
|
|
try:
|
|
plugin_class = ep.load()
|
|
plugin = plugin_class()
|
|
self.registry.register(plugin)
|
|
plugins.append(plugin)
|
|
except Exception as e:
|
|
print(f"Warning: Failed to load plugin {ep.name}: {e}")
|
|
except Exception as e:
|
|
print(f"Warning: Could not load entry points: {e}")
|
|
return plugins
|
|
|
|
def load_plugin_module(self, module_name: str) -> Plugin:
|
|
module = __import__(module_name, fromlist=[''])
|
|
if not hasattr(module, 'register'):
|
|
raise ValueError(f"Module {module_name} does not have a 'register' function")
|
|
|
|
plugin = module.register()
|
|
if not isinstance(plugin, Plugin):
|
|
raise ValueError("register() must return a Plugin instance")
|
|
|
|
self.registry.register(plugin)
|
|
return plugin
|
|
|
|
def load_from_path(self, path: str) -> List[Plugin]:
|
|
import importlib.util
|
|
|
|
plugins = []
|
|
import os
|
|
if os.path.isdir(path):
|
|
for filename in os.listdir(path):
|
|
if filename.endswith('.py') and not filename.startswith('_'):
|
|
module_name = filename[:-3]
|
|
file_path = os.path.join(path, filename)
|
|
spec = importlib.util.spec_from_file_location(module_name, file_path)
|
|
if spec and spec.loader:
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
if hasattr(module, 'register'):
|
|
plugin = module.register()
|
|
if isinstance(plugin, Plugin):
|
|
self.registry.register(plugin)
|
|
plugins.append(plugin)
|
|
return plugins
|
|
|
|
def discover_and_load(self) -> List[Plugin]:
|
|
plugins = self.load_entry_points()
|
|
return plugins
|
|
|
|
def get_registry(self) -> PluginRegistry:
|
|
return self.registry
|