Add PluginManager class and move logic from main => PluginManager or Plugin as appropriate

examples

@@ -1 +1 @@
-Subproject commit f8b077f3764c16935462ffb818bdb5aeda75222b
+Subproject commit 1466438938c98442aa6de55065b7e6a06a7e8d50

src/sshare/logger.py

@@ -12,12 +12,14 @@
 # You should have received a copy of the GNU General Public License along with
 # SSHare. If not, see <https://www.gnu.org/licenses/>.
 
+import sys
+
 from sshare.plugin import Plugin
 
 class Logger:
     def __init__(self, *args, **kwargs):
-        if kwargs.get("preload_command_line") == True:
+        if kwargs.get("command_line"):
-            self._loggers = [ Plugin.internal("command_line") ]
+            self._loggers = [ kwargs["command_line"] ]
         else:
             self._loggers = []
         self.add(*args)

src/sshare/main.py

@@ -15,7 +15,6 @@
 import argparse
 import getpass
 import os
-import os.path
 import time
 import tomllib
 import subprocess
@@ -24,175 +23,79 @@ from pathlib import Path
 
 from sshare.logger         import Logger
 from sshare.plugin         import Plugin
-from sshare.plugins.config import Flag
+from sshare.plugin         import PluginManager
-from sshare.plugins.config import NoArgument
-from sshare.plugins.config import NoDefault
 from sshare.version        import version
 
 
 def main():
-    config_directory = Path(os.environ.get("XDG_CONFIG_DIR") or f"{os.environ["HOME"]}/.config") / "sshare"
+    # TODO: Add --config flag
+    config_directory = Path(os.environ.get("XDG_CONFIG_DIR", f"{os.environ["HOME"]}/.config")) / "sshare"
     with open(config_directory / "config.toml", mode="rb") as file:
         config = tomllib.load(file)
 
+    INTERNAL_PLUGIN_LOCATION = "sshare.plugins.default"
     # Load command line early and set it as the active logger
     # so that it can be used to report errors while loading and
-    # configuring plugins
+    # configuring other loggers
-    # i.e. before other logging plugins have had a chance to be initialised
+    command_line = Plugin.internal(INTERNAL_PLUGIN_LOCATION, "command_line", config.get("plugins", dict()))
-    logger = Logger(preload_command_line=True)
+    logger = Logger(command_line=command_line)
-
+    arg_parser = argparse.ArgumentParser(
-    # Load inbuilt plugins
-    plugins_flat = [
-        Plugin.internal("file"),
-        Plugin.internal("current_time"),
-        Plugin.internal("append_type"),
-        Plugin.internal("ssh"),
-        Plugin.internal("log_result"),
-    ]
-    plugins = {}
-    for type in [ "logger", "source", "name", "upload", "result" ]:
-        plugins[type] = { "active": [], "inactive": [] }
-
-    # Load external plugins
-    sys.dont_write_bytecode = True
-    for path in (config_directory / "plugins").iterdir():
-        if path.is_file() and path.suffix == ".py":
-            plugins_flat.append(Plugin.external(path))
-    sys.dont_write_bytecode = False
-
-    # Set plugin configurations from config file
-    # Load plugin arguments and detect conflicts
-    error = False
-    argument_map = {}
-    used_arguments = {}
-    parser = argparse.ArgumentParser(
         prog = "sshare",
         description = "Upload files to a server via ssh",
     )
-    parser.add_argument(
+    arg_parser.add_argument(
         "-v",
         "--version",
         action="version",
         version=f"%(prog)s version {version}",
     )
-    if config.get("plugins") == None:
+    plugins = PluginManager(
-        config["plugins"] = {}
+        [ "logger", "source", "name", "upload", "result" ],
-    for plugin in plugins_flat:
+        logger,
-        if hasattr(plugin, "config"):
+        config.get("plugins", dict()),
-            plugin_config = config["plugins"].get(plugin.name)
+        arg_parser,
-            if plugin_config != None:
+    )
-                for config_entry in plugin_config.items():
+    plugins.add_from(
-                    plugin.config[config_entry[0]] = config_entry[1]
+        Plugin.internal(INTERNAL_PLUGIN_LOCATION),
-        else:
+        "file",
-            setattr(plugin, "config", {})
+        "current_time",
-        if hasattr(plugin, "args"):
+        "append_type",
-            for arg_name, arg in plugin.args.items():
+        "ssh",
-                if arg.is_valid():
+        "log_result",
-                    arg.bind(plugin, arg_name)
+    )
-                    def check_flag(flag):
+    sys.dont_write_bytecode = True
-                        if flag in used_arguments:
+    plugins.add_from(
-                            logger.error(f"Error: Argument '{arg_name}' for plugin '{plugin.name}' has conflict. Flag '{flag}' is also used by plugin '{used_arguments[arg.short]}'")
+        Plugin.external,
-                            error = True
+        *[
-                    check_flag(arg.short)
+            path for
-                    check_flag(arg.long)
+            path in
-                    arg.add(parser, used_arguments)
+            (config_directory / "plugins").iterdir()
-                    argument_map[arg.dest] = plugin, arg_name
+            if path.is_file() and path.suffix == ".py"
-                else:
+        ]
-                    logger.error(f"Error: Argument '{arg_name}' must set either one or both of short and long flag parameters")
+    )
-                    error = True
+    sys.dont_write_bytecode = False
-    if error:
+    plugins.activate("logger")
-        sys.exit(1)
+    logger.add(*plugins.logger.active)
-
+    plugins.activate()
-    arguments = parser.parse_args()
-    for arg, (plugin, config) in list(argument_map.items()):
-        value = getattr(arguments, arg)
-        if value != NoArgument:
-            if value != Flag:
-                plugin.config[config] = value
-            del argument_map[arg]
-
-    # Sort plugins by type and check activation criteria
-    error = False
-    for plugin in plugins_flat:
-        if isinstance(plugin.plugin_type, str):
-            plugin.plugin_type = [ plugin.plugin_type ]
-        for plugin_type in plugin.plugin_type:
-            plugins_of_type = plugins.get(plugin_type)
-            if plugins_of_type == None:
-                logger.error(f"Error: Plugin '{plugin.name}' has an invalid plugin type '{plugin_type}'")
-                error = True
-            else:
-                active = True
-                if hasattr(plugin, "activate"):
-                    criteria = plugin.activate
-                    if isinstance(plugin.activate, dict):
-                        criteria = plugin.activate.get(plugin_type)
-                    if criteria != None:
-                        for criterion in criteria:
-                            active = not plugin.args[criterion].dest in argument_map
-                            if not active:
-                                break
-                plugins_of_type["active" if active else "inactive"].append(plugin)
-                if active:
-                    for config_entry in plugin.config.items():
-                        if config_entry[1] == NoDefault:
-                            logger.error(f"Error: Value 'plugins.{plugin.name}.{config_entry[0]}' has no default value and must be specified explicitly")
-                            error = True
-    for plugin_type, plugins_of_type in plugins.items():
-        if len(plugins_of_type["active"]) == 0 and plugin_type != "logger":
-            if len(plugins_of_type["inactive"]) == 0:
-                logger.error(f"No '{plugin_type}' plugins available. Atleast one must be provided")
-            else:
-                logger.error(f"No '{plugin_type}' plugins activated. Activate at least one of:")
-                for plugin in plugins_of_type["inactive"]:
-                    logger.error(f"{plugin.name}:")
-                    criteria = plugin.activate
-                    if isinstance(plugin.activate, dict):
-                        criteria = plugin.activate[plugin_type]
-                    for criterion in criteria:
-                        logger.error(f"  {plugin.args[criterion]}")
-            error = True
-    if error:
-        sys.exit(1)
-
-    # Objectify configs
-    error = False
-    class PluginConfig: pass
-    for plugin in plugins_flat:
-        config = plugin.config
-        plugin.config = PluginConfig()
-        for config_entry in config.items():
-            setattr(plugin.config, config_entry[0], config_entry[1])
-    if error:
-        sys.exit(1)
-
-    # Initialise plugins
-    for plugin in plugins_flat:
-        setattr(plugin, "logger", logger)
-        if hasattr(plugin, "init"):
-            error = error or plugin.init()
-
-    logger.add(*plugins["logger"]["active"])
 
     sources = []
-    for plugin in plugins["source"]["active"]:
+    for plugin in plugins.source.active:
         sources.append(plugin.get_source())
     if len(sources) == 0:
-        logger.error("Error: No sources provided. Must activate at least one source plugin")
+        logger.error("Error: No sources provided. Must activate at least one source plugin.")
-        log_activations(logger, plugins["source"])
 
     for index, source in enumerate(sources):
         name = ""
-        for plugin in plugins["name"]["active"]:
+        for plugin in plugins.name.active:
             name = plugin.get_name(name, source)
         sources[index] = name, source
 
     for name, source in sources:
-        for plugin in plugins["upload"]["active"]:
+        for plugin in plugins.upload.active:
             plugin.upload(name, source)
 
     for name, _ in sources:
-        for plugin in plugins["result"]["active"]:
+        for plugin in plugins.result.active:
             plugin.result(name)
 
     sys.exit(0)

src/sshare/plugin.py

@@ -15,21 +15,127 @@
 import importlib
 import importlib.util
 
+from sshare.plugins.config import Flag
+from sshare.plugins.config import NoDefault
+
+class PluginManager:
+    def __init__(self, types, logger, config, arg_parser):
+        self._uninitialized = []
+        self._logger        = logger
+        self._config        = config
+        self._arg_parser    = arg_parser
+
+        class PluginState:
+            def __init__(self):
+                self.active = []
+                self.inactive = []
+        for type in types:
+            setattr(self, type, PluginState())
+
+    def add_from(self, location, *args, **kwargs):
+        for plugin in args:
+            plugin = location(plugin, self._config)
+            plugin.set_logger(self._logger)
+            plugin.add_args(self._arg_parser)
+            self._uninitialized.append(plugin)
+
+    def activate(self, activate_type=None):
+        args = self._arg_parser.parse_args()
+        for plugin in self._uninitialized.copy():
+            if activate_type == None or activate_type in plugin.plugin_type:
+                self._uninitialized.remove(plugin)
+                active = plugin.load_args_and_activate(args)
+                for type in plugin.plugin_type:
+                    getattr(
+                        getattr(self, type),
+                        active[type],
+                    ).append(plugin)
+
 class Plugin:
-    def __init__(self, name, module):
+    def __init__(self, name, module, external_config):
         self.__dict__ = module.__dict__
         self.name     = name
 
-    @staticmethod
+        if not isinstance(self.plugin_type, set):
-    def internal(name):
+            self.plugin_type = { self.plugin_type }
-        return Plugin(name, importlib.import_module(f"sshare.plugins.default.{name}"))
+
+        if hasattr(self, "activate"):
+            if not isinstance(self.activate, dict):
+                activate = self.activate
+                self.activate = dict()
+                for plugin_type in self.plugin_type:
+                    self.activate[plugin_type] = activate
+        else:
+            self.activate = dict()
+            for plugin_type in self.plugin_type:
+                self.activate[plugin_type] = set()
+
+        if hasattr(self, "config"):
+            config = self.config
+        else:
+            config = dict()
+        if external_config == None:
+            external_config = dict()
+        for key in config.keys():
+            if key in external_config:
+                config[key] = external_config[key]
+
+        if hasattr(self, "args"):
+            for arg in self.args.items():
+                arg[1].bind(self, arg[0])
+                value = arg[1].default()
+                if value != NoDefault:
+                    config[arg[0]] = value
+        else:
+            self.args = dict()
+
+        class Config: pass
+        flat_config = Config()
+        flat_config.__dict__ = config
+        self.config = flat_config
+
+    def set_logger(self, logger):
+        self.logger = logger
+
+    def add_args(self, arg_parser):
+        for arg in self.args.values():
+            arg.add(arg_parser)
+
+    def load_args_and_activate(self, args):
+        passed_args = set()
+        for arg_name, arg in self.args.items():
+            was_set, value = arg.extract(args)
+            if was_set:
+                if value != Flag:
+                    setattr(self.config, arg_name, value)
+                passed_args.add(arg_name)
+        activate = dict()
+        run_init = False
+        for type in self.plugin_type:
+            if self.activate[type] <= passed_args:
+                activate[type] = "active"
+                run_init = True
+            else:
+                activate[type] = "inactive"
+        if run_init and hasattr(self, "init"):
+            self.init()
+        return activate
     
     @staticmethod
-    def external(path):
+    def internal(location, name=None, config=None):
+        def _load_internal(_name, _config):
+            return Plugin(_name, importlib.import_module(f"{location}.{_name}"), _config.get(_name, dict()))
+        if name == None:
+            return _load_internal
+        else:
+            return _load_internal(name, config)
+
+    @staticmethod
+    def external(path, config):
         module_spec = importlib.util.spec_from_file_location(
             path.stem,
             path.as_posix(),
         )
         module = importlib.util.module_from_spec(module_spec)
         module_spec.loader.exec_module(module)
-        return Plugin(path.stem, module)
+        return Plugin(path.stem, module, config.get(path.stem, dict()))

src/sshare/plugins/config.py

@@ -14,71 +14,64 @@
 
 class NoDefault: pass
 
-class NoArgument: pass
+def Flag(short=None, long=None, help=None):
-class Flag: pass
+    return Argument(
+        short,
+        long,
+        action="store_const",
+        const=Flag,
+        default=False,
+        help=help,
+    )
 
 class Argument:
-    def __init__(self,
+    def __init__(self, short=None, long=None, **kwargs):
-                 short=None,
+        class _None:
-                 long=None,
+            def __init__(self, default):
-                 action=None,
+                self.default = default
-                 nargs=None,
+        self._None = _None
-                 const=None,
-                 default=NoArgument,
-                 type=None,
-                 choices=None,
-                 required=None,
-                 help=None):
-        self.short   = short
-        self.long    = long
-        self.action  = action
-        self.nargs   = nargs
-        self.const   = const
-        self.default = default
-        self.type    = type
-        self.choices = choices
-        self.help    = help
 
-    def __str__(self):
+        self.short = short
-        if self.short and self.long:
+        self.long  = long
-            pretty = f"-{self.short}, --{self.long}"
-        elif self.long:
-            pretty = f"--{self.long}"
-        else:
-            pretty = f"-{self.short}"
-        return pretty + f" {self.help}"
 
-    def is_valid(self):
+        if not "default" in kwargs:
-        return (self.short != None and self.short != "") or (self.long != None and self.long != "")
+            kwargs["default"] = NoDefault
+        kwargs["default"] = _None(kwargs["default"])
+        self._kwargs = kwargs
 
     def bind(self, plugin, argument):
-        self.plugin  = plugin
+        self._plugin   = plugin.name
-        self.metavar = argument
+        self._argument = argument
-        self.dest    = f"{plugin.name}_{argument}"
 
-    def add(self, parser, used_arguments):
+    def default(self):
-        keywords = [
+        value = self._kwargs["default"]
-            "action",
+        if isinstance(value, self._None):
-            "nargs",
+            value = value.default
-            "const",
+        return value
-            "default",
+
-            "type",
+    def dest(self):
-            "choices",
+        return f"{self._plugin}_{self._argument}"
-            "help",
+
-            "metavar",
+    def extract(self, arguments):
-            "dest"
+        value = getattr(arguments, self.dest())
-        ]
+        was_set = True
-        kwargs = {}
+        if isinstance(value, self._None):
-        for keyword in keywords:
+            was_set = False
-            value = getattr(self, keyword)
+            value = value.default
-            if value != None:
+        return was_set, value
-                kwargs[keyword] = value
+
-        parser.add_argument(
+    def add(self, arg_parser):
-            f"-{self.short}",
+        flags = []
-            f"--{self.long}",
+        if self.short:
+            flags.append(f"-{self.short}")
+        long = self.long or self._argument
+        if long:
+            flags.append(f"--{long}")
+        kwargs = self._kwargs | {
+            "metavar": self._argument,
+            "dest": self.dest()
+        }
+        arg_parser.add_argument(
+            *flags,
             **kwargs
         )
-        if self.short:
-            used_arguments["short"] = self.plugin
-        if self.long:
-            used_arguments["long"] = self.plugin

src/sshare/plugins/default/file.py

@@ -18,11 +18,10 @@ from ..source import File
 
 plugin_type = "source"
 
-activate = [ "file" ]
+activate = { "file" }
 args = {
     "file": Argument(
         short="f",
-        long="file",
         help="Upload a file"
     )
 }