from types import ModuleType from typing import Dict, Optional from inspect import isclass from pkgutil import iter_modules from pathlib import Path from os.path import dirname from importlib import import_module from inspect import signature from raven.pluginbase import PluginBase class PluginRegistry(): plugins: Dict[str, PluginBase] = {} modules: [ModuleType] = [] @staticmethod def _register_plugin(name: str, plugin: PluginBase): PluginRegistry.plugins[name] = plugin # @staticmethod # def load_module(module_name: str) -> ModuleType: # module_name = f"raven.plugins.{module_name}" # module = import_module(module_name) # PluginRegistry.modules.append(module) # return module @staticmethod def load_plugin(plugin_name: str) -> PluginBase: module_name = f"raven.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._register_plugin(plugin_name, plugin_class()) return plugin_class raise RuntimeError("plugin %s not found" % plugin_name) # @staticmethod # def get_modules() -> [ModuleType]: # return PluginRegistry.modules.copy() @staticmethod def get_plugins() -> [PluginBase]: return PluginRegistry.modules.copy() @staticmethod def execute_single(function_name: str, *args) -> Optional[any]: return PluginRegistry._execute(function_name, True, *args) @staticmethod def execute(function_name: str, *args) -> [any]: return PluginRegistry._execute(function_name, False, *args) @staticmethod def _execute(function_name: str, return_first: bool, *args) -> []: result = [] for plugin in PluginRegistry.plugins.values(): fun = getattr(plugin, function_name, None) if callable(fun): sig = signature(fun) if len(sig.parameters) != len(args): raise RuntimeError("method %s.%s has wrong number of arguments. expected %s but was %s " % ( plugin, function_name, len(args), len(sig.parameters))) print("calling %s with args %s" % (fun, args)) if len(args) == 0: return_value = fun() elif len(args) == 1: return_value = fun(args[0]) elif len(args) == 2: return_value = fun(args[0], args[1]) elif len(args) == 3: return_value = fun(args[0], args[1], args[2]) elif len(args) == 4: return_value = fun(args[0], args[1], args[2], args[3]) elif len(args) == 5: return_value = fun(args[0], args[1], args[2], args[3], args[4]) else: raise Exception("too many arguments") if return_first: return return_value result.append(return_value) if return_first: return None return result