from types import ModuleType from typing import Dict, Optional from inspect import isclass from importlib import import_module from inspect import signature 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)