add ZonedPluginRegistry

src/pluginregistry.py

@@ -4,75 +4,28 @@ 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: Dict[str, PluginBase] = {}
+    plugins = ZonedPluginRegistry({})
-
-    modules: [ModuleType] = []
 
     @staticmethod
-    def _register_plugin(name: str, plugin: PluginBase):
+    def load_plugin(plugin_name: str):
-        PluginRegistry.plugins[name] = plugin
-
-    @staticmethod
-    def load_plugin(plugin_name: str) -> PluginBase:
         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._register_plugin(plugin_name, plugin_class())
+                PluginRegistry.plugins.register_plugin(plugin_name, plugin_class())
-                return plugin_class
+                return
         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)
+        return PluginRegistry.plugins.execute_single(function_name, *args)
 
     @staticmethod
     def execute(function_name: str, *args) -> [any]:
-        return PluginRegistry._execute(function_name, False, *args)
+        return PluginRegistry.plugins.execute(function_name, *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

src/zonedpluginregistry.py

@@ -0,0 +1,52 @@
+from inspect import signature
+from typing import Dict, Optional
+
+from src.pluginbase import PluginBase
+
+
+class ZonedPluginRegistry:
+
+    def __init__(self, plugins: Dict[str, PluginBase]):
+        self.plugins: Dict[str, PluginBase] = plugins
+
+    def register_plugin(self, name: str, plugin: PluginBase):
+        self.plugins[name] = plugin
+
+    def execute_single(self, function_name: str, *args) -> Optional[any]:
+        return self._execute(function_name, True, *args)
+
+    def execute(self, function_name: str, *args) -> [any]:
+        return self._execute(function_name, False, *args)
+
+    def _execute(self, function_name: str, return_first: bool, *args) -> []:
+        result = []
+        for plugin in self.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