""" Helpers for plugin app """ import os import subprocess import pathlib import sysconfig import traceback import inspect import pkgutil from django.conf import settings from django.core.exceptions import AppRegistryNotReady # region logging / errors class IntegrationPluginError(Exception): """ Error that encapsulates another error and adds the path / reference of the raising plugin """ def __init__(self, path, message): self.path = path self.message = message def __str__(self): return self.message class MixinImplementationError(ValueError): """ Error if mixin was implemented wrong in plugin Mostly raised if constant is missing """ pass class MixinNotImplementedError(NotImplementedError): """ Error if necessary mixin function was not overwritten """ pass def log_error(error, reference: str = 'general'): """ Log an plugin error """ from plugin import registry # make sure the registry is set up if reference not in registry.errors: registry.errors[reference] = [] # add error to stack registry.errors[reference].append(error) def handle_error(error, do_raise: bool = True, do_log: bool = True, do_return: bool = False, log_name: str = ''): """ Handles an error and casts it as an IntegrationPluginError """ package_path = traceback.extract_tb(error.__traceback__)[-1].filename install_path = sysconfig.get_paths()["purelib"] try: package_name = pathlib.Path(package_path).relative_to(install_path).parts[0] except ValueError: # is file - loaded -> form a name for that path_obj = pathlib.Path(package_path).relative_to(settings.BASE_DIR) path_parts = [*path_obj.parts] path_parts[-1] = path_parts[-1].replace(path_obj.suffix, '') # remove suffix # remove path preixes if path_parts[0] == 'plugin': path_parts.remove('plugin') path_parts.pop(0) else: path_parts.remove('plugins') package_name = '.'.join(path_parts) if do_log: log_kwargs = {} if log_name: log_kwargs['reference'] = log_name log_error({package_name: str(error)}, **log_kwargs) new_error = IntegrationPluginError(package_name, str(error)) if do_raise: raise IntegrationPluginError(package_name, str(error)) if do_return: return new_error # endregion # region git-helpers def get_git_log(path): """ Get dict with info of the last commit to file named in path """ path = path.replace(os.path.dirname(settings.BASE_DIR), '')[1:] command = ['git', 'log', '-n', '1', "--pretty=format:'%H%n%aN%n%aE%n%aI%n%f%n%G?%n%GK'", '--follow', '--', path] try: output = str(subprocess.check_output(command, cwd=os.path.dirname(settings.BASE_DIR)), 'utf-8')[1:-1] if output: output = output.split('\n') else: output = 7 * [''] except subprocess.CalledProcessError: output = 7 * [''] return {'hash': output[0], 'author': output[1], 'mail': output[2], 'date': output[3], 'message': output[4], 'verified': output[5], 'key': output[6]} class GitStatus: """ Class for resolving git gpg singing state """ class Definition: """ Definition of a git gpg sing state """ key: str = 'N' status: int = 2 msg: str = '' def __init__(self, key: str = 'N', status: int = 2, msg: str = '') -> None: self.key = key self.status = status self.msg = msg N = Definition(key='N', status=2, msg='no signature',) G = Definition(key='G', status=0, msg='valid signature',) B = Definition(key='B', status=2, msg='bad signature',) U = Definition(key='U', status=1, msg='good signature, unknown validity',) X = Definition(key='X', status=1, msg='good signature, expired',) Y = Definition(key='Y', status=1, msg='good signature, expired key',) R = Definition(key='R', status=2, msg='good signature, revoked key',) E = Definition(key='E', status=1, msg='cannot be checked',) # endregion # region plugin finders def get_modules(pkg): """get all modules in a package""" context = {} for loader, name, ispkg in pkgutil.walk_packages(pkg.__path__): try: module = loader.find_module(name).load_module(name) pkg_names = getattr(module, '__all__', None) for k, v in vars(module).items(): if not k.startswith('_') and (pkg_names is None or k in pkg_names): context[k] = v context[name] = module except AppRegistryNotReady: pass except Exception as error: # this 'protects' against malformed plugin modules by more or less silently failing # log to stack log_error({name: str(error)}, 'discovery') return [v for k, v in context.items()] def get_classes(module): """get all classes in a given module""" return inspect.getmembers(module, inspect.isclass) def get_plugins(pkg, baseclass): """ Return a list of all modules under a given package. - Modules must be a subclass of the provided 'baseclass' - Modules must have a non-empty PLUGIN_NAME parameter """ plugins = [] modules = get_modules(pkg) # Iterate through each module in the package for mod in modules: # Iterate through each class in the module for item in get_classes(mod): plugin = item[1] if issubclass(plugin, baseclass) and plugin.PLUGIN_NAME: plugins.append(plugin) return plugins # endregion