from types import ModuleType from typing import Dict 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 get_plugins_by_function(function_name: str) -> [PluginBase]: result = [] for plugin in PluginRegistry.plugins.values(): fun = getattr(plugin, function_name, None) if callable(fun): result.append(plugin) return result @staticmethod def load_module(module_name: str) -> ModuleType: module_name = f"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"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(function_name: str, *args): for plugin in PluginRegistry.plugins.values(): fun = getattr(plugin, function_name, None) sig = signature(fun) if callable(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))) fun(args)