src.extra.plugins_loader

  1"""
  2Loads all plugins
  3"""
  4
  5import importlib
  6import inspect
  7import logging
  8import pkgutil
  9import sys
 10from sys import stdout
 11from typing import Any
 12
 13import src.constants as cst
 14from src.cmd_types.commands import ExecutableCommand, UndoableCommand
 15from src.cmd_types.meta import CommandMetadata
 16from src.cmd_types.plugins import PluginMetadata
 17
 18RESTRICTED = (ExecutableCommand, UndoableCommand)
 19
 20class PluginLoader:
 21    """
 22    Class to load plugins
 23
 24    :param pkg_dir: dir with plugins
 25    :type pkg_dir: str
 26
 27    :param prefix: prefix of plugins files
 28    :type prefix: str
 29
 30    :param strict: Whether raise an error on import or not
 31    :type strict: bool
 32
 33    :raise ImportError: if plugin cannot be loaded. Rather the command was already loaded or plugins has some errors on import
 34
 35    """
 36    def __init__(self, pkg_dir: str = cst.PLUGINS_DIR, prefix: str = cst.PLUGINS_PREFIX, strict: bool = cst.STRICT_PLUGIN_LOADING):
 37        self.pkg_dir = pkg_dir
 38        self.prefix = prefix
 39        self.logger = logging.Logger(__name__)
 40
 41        self.commands: dict[str, CommandMetadata] = {}
 42        """dict of loaded commands"""
 43
 44        self.strict = strict
 45        self.__init_logger()
 46        self.non_default: dict[str, PluginMetadata] = {}
 47        """dict of non-default plugins(will be loaded only after default ones)"""
 48
 49    def __init_logger(self):
 50        handlers = [
 51            logging.FileHandler(cst.LOG_FILE, mode="a", encoding="utf-8"),
 52            logging.StreamHandler(stdout),
 53        ]
 54        formatter = logging.Formatter(cst.FORMAT_LOADER)
 55        for handler in handlers:
 56            handler.setFormatter(formatter)
 57            self.logger.addHandler(handler)
 58        self.logger.setLevel(cst.LOGGING_LEVEL)
 59
 60    def load_plugins(self):
 61        """
 62        Loads all plugins
 63        :return:
 64        """
 65        plugins_pkg = importlib.import_module(self.pkg_dir)
 66        for importer, module_name, is_pkg in pkgutil.iter_modules(plugins_pkg.__path__):
 67            if module_name.startswith(self.prefix) and not is_pkg:
 68                self._load_module(module_name)
 69        for k in self.non_default.keys():
 70            self._load_module(k, False)
 71
 72    def warn_or_error(self, *, warn_msg: str = "", exc: Any = ImportError):
 73        """
 74        Will warn if strict set to False else will raise given exception
 75        :param warn_msg: message for warning
 76        :param exc: exception to raise
 77        """
 78        if not self.strict:
 79            self.logger.warning(warn_msg)
 80
 81        else:
 82            raise exc
 83
 84
 85    def _load_module(self, module_name: str, defaults: bool = True):
 86        """
 87        Imports one module.
 88        :param module_name: name of the module to import
 89        :param defaults: if set to True, will only load if it is non-default. Otherwise, will load commands from self.non_default
 90        :return:
 91        """
 92        full_module_name = defaults * f'{self.pkg_dir}.' + f"{module_name}"
 93        if defaults:
 94            try:
 95                if full_module_name in sys.modules:
 96                    module = sys.modules[full_module_name]
 97                    importlib.reload(module)
 98                    self.logger.debug(f"Module {module.__name__} already imported: reloading")
 99                else:
100                    module = importlib.import_module(full_module_name)
101            except Exception as e:
102                self.warn_or_error(warn_msg=f"Failed to load module {full_module_name}: {e}", exc=e)
103                return
104            author = getattr(module, "__author__", None)
105            version = getattr(module, "__version__", None)
106            if author != "default" :
107                self.non_default[full_module_name] = PluginMetadata(module = module, author = author, version = version)
108                return
109        else:
110            obj_import = self.non_default[full_module_name]
111            module = obj_import.module
112            author = obj_import.author
113            version = obj_import.version
114        self.logger.debug(f"Loading {'non-default' if not defaults else 'default'} module {full_module_name}...")
115        for name, obj in inspect.getmembers(module):
116            if inspect.isclass(obj):
117                if issubclass(obj, ExecutableCommand) and obj not in RESTRICTED and getattr(obj, "name", None):
118                    cmd_name = obj.name
119                    if self.commands.get(cmd_name):
120
121                        warn_msg = f"Command {cmd_name} in module {full_module_name} already exists, skipping"
122
123                        exc = ImportError(f"{cmd_name} imported twice", name = module_name, path = full_module_name)
124
125                        self.warn_or_error(warn_msg=warn_msg, exc=exc)
126
127                    elif " " in cmd_name:
128                        warn_msg = f"Command {cmd_name} in module {full_module_name} has spaces in it"
129                        exc = ImportError(warn_msg, name = module_name, path = full_module_name)
130                        warn_msg+=", skipping"
131
132                        self.warn_or_error(warn_msg=warn_msg, exc=exc)
133
134                    else:
135                        self.logger.debug(f"Loading command {cmd_name}...")
136                        self.commands[cmd_name] = CommandMetadata(
137                            name = cmd_name,
138                            plugin_name = module_name,
139                            plugin_author=author,
140                            plugin_version=version,
141                            cmd = obj
142                            )
143                        self.logger.debug(f"Command {cmd_name} in module {full_module_name} loaded")
144
145        self.logger.debug(f"Loaded module {full_module_name}")