89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
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 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 executeSingle(function_name: str, *args):
|
|
return PluginRegistry._execute(function_name, True, *args)
|
|
|
|
@staticmethod
|
|
def execute(function_name: str, *args) -> []:
|
|
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
|