38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
from typing import Dict, Optional
|
|
from inspect import isclass
|
|
from importlib import import_module
|
|
from src.pluginbase import PluginBase
|
|
from src.zonedpluginregistry import ZonedPluginRegistry
|
|
|
|
|
|
class PluginRegistry():
|
|
plugins = ZonedPluginRegistry()
|
|
|
|
@staticmethod
|
|
def load_plugin(plugin_name: str):
|
|
module_name = f"src.plugins.{plugin_name.lower()}"
|
|
module = import_module(module_name)
|
|
if plugin_name in dir(module):
|
|
plugin_class = getattr(module, plugin_name)
|
|
|
|
if isclass(plugin_class) and issubclass(plugin_class, PluginBase):
|
|
PluginRegistry.plugins.register_plugin(plugin_name, plugin_class())
|
|
return
|
|
raise RuntimeError("plugin %s not found" % plugin_name)
|
|
|
|
@staticmethod
|
|
def create_zoned_plugin_registry():
|
|
return PluginRegistry.plugins.copy()
|
|
|
|
@staticmethod
|
|
def execute_single(function_name: str, *args) -> Optional[any]:
|
|
return PluginRegistry.plugins.execute_single(function_name, *args)
|
|
|
|
@staticmethod
|
|
def execute(function_name: str, *args) -> [any]:
|
|
return PluginRegistry.plugins.execute(function_name, *args)
|
|
|
|
@staticmethod
|
|
def execute_flat_map(function_name: str, *args) -> [any]:
|
|
return PluginRegistry.plugins.execute_flat_map(function_name, *args)
|