From de59bbf2be1008f723a579f186f6ce15a347842f Mon Sep 17 00:00:00 2001 From: Matthias Mair Date: Mon, 13 Feb 2023 21:19:51 +0100 Subject: [PATCH] move loaded/unloaded mixins out into seperate modules --- InvenTree/plugin/base/integration/AppMixin.py | 192 +++++++ .../plugin/base/integration/ScheduleMixin.py | 211 +++++++ .../plugin/base/integration/SettingsMixin.py | 73 +++ .../plugin/base/integration/UrlsMixin.py | 71 +++ InvenTree/plugin/base/integration/mixins.py | 524 +----------------- InvenTree/plugin/mixins/__init__.py | 11 +- InvenTree/plugin/registry.py | 6 +- 7 files changed, 560 insertions(+), 528 deletions(-) create mode 100644 InvenTree/plugin/base/integration/AppMixin.py create mode 100644 InvenTree/plugin/base/integration/ScheduleMixin.py create mode 100644 InvenTree/plugin/base/integration/SettingsMixin.py create mode 100644 InvenTree/plugin/base/integration/UrlsMixin.py diff --git a/InvenTree/plugin/base/integration/AppMixin.py b/InvenTree/plugin/base/integration/AppMixin.py new file mode 100644 index 0000000000..b3e4e83f59 --- /dev/null +++ b/InvenTree/plugin/base/integration/AppMixin.py @@ -0,0 +1,192 @@ +"""Plugin mixin class for AppMixin.""" +import logging +from importlib import reload +from typing import OrderedDict + +from django.apps import apps +from django.conf import settings +from django.contrib import admin + +from plugin.helpers import handle_error + +logger = logging.getLogger('inventree') + + +class AppMixin: + """Mixin that enables full django app functions for a plugin.""" + + class MixinMeta: + """Meta options for this mixin.""" + + MIXIN_NAME = 'App registration' + + def __init__(self): + """Register mixin.""" + super().__init__() + self.add_mixin('app', 'has_app', __class__) + + def _activate_mixin(self, plugins, force_reload=False, full_reload: bool = False): + """Activate AppMixin plugins - add custom apps and reload. + + Args: + plugins (dict): List of IntegrationPlugins that should be installed + force_reload (bool, optional): Only reload base apps. Defaults to False. + full_reload (bool, optional): Reload everything - including plugin mechanism. Defaults to False. + """ + from common.models import InvenTreeSetting + + if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_APP'): + logger.info('Registering IntegrationPlugin apps') + apps_changed = False + + # add them to the INSTALLED_APPS + for _key, plugin in plugins: + if plugin.mixin_enabled('app'): + plugin_path = self._get_plugin_path(plugin) + if plugin_path not in settings.INSTALLED_APPS: + settings.INSTALLED_APPS += [plugin_path] + self.installed_apps += [plugin_path] + apps_changed = True + # if apps were changed or force loading base apps -> reload + if apps_changed or force_reload: + # first startup or force loading of base apps -> registry is prob false + if self.apps_loading or force_reload: + self.apps_loading = False + self._reload_apps(force_reload=True, full_reload=full_reload) + else: + self._reload_apps(full_reload=full_reload) + + # rediscover models/ admin sites + self._reregister_contrib_apps() + + # update urls - must be last as models must be registered for creating admin routes + self._update_urls() + + def _deactivate_mixin(self): + """Deactivate AppMixin plugins - some magic required.""" + # unregister models from admin + for plugin_path in self.installed_apps: + models = [] # the modelrefs need to be collected as poping an item in a iter is not welcomed + app_name = plugin_path.split('.')[-1] + try: + app_config = apps.get_app_config(app_name) + + # check all models + for model in app_config.get_models(): + # remove model from admin site + try: + admin.site.unregister(model) + except Exception: # pragma: no cover + pass + models += [model._meta.model_name] + except LookupError: # pragma: no cover + # if an error occurs the app was never loaded right -> so nothing to do anymore + logger.debug(f'{app_name} App was not found during deregistering') + break + + # unregister the models (yes, models are just kept in multilevel dicts) + for model in models: + # remove model from general registry + apps.all_models[plugin_path].pop(model) + + # clear the registry for that app + # so that the import trick will work on reloading the same plugin + # -> the registry is kept for the whole lifecycle + if models and app_name in apps.all_models: + apps.all_models.pop(app_name) + + # remove plugin from installed_apps + self._clean_installed_apps() + + # reset load flag and reload apps + settings.INTEGRATION_APPS_LOADED = False + self._reload_apps() + + # update urls to remove the apps from the site admin + self._update_urls() + + # region helpers + def _reregister_contrib_apps(self): + """Fix reloading of contrib apps - models and admin. + + This is needed if plugins were loaded earlier and then reloaded as models and admins rely on imports. + Those register models and admin in their respective objects (e.g. admin.site for admin). + """ + for plugin_path in self.installed_apps: + try: + app_name = plugin_path.split('.')[-1] + app_config = apps.get_app_config(app_name) + except LookupError: # pragma: no cover + # the plugin was never loaded correctly + logger.debug(f'{app_name} App was not found during deregistering') + break + + # reload models if they were set + # models_module gets set if models were defined - even after multiple loads + # on a reload the models registery is empty but models_module is not + if app_config.models_module and len(app_config.models) == 0: + reload(app_config.models_module) + + # check for all models if they are registered with the site admin + model_not_reg = False + for model in app_config.get_models(): + if not admin.site.is_registered(model): + model_not_reg = True + + # reload admin if at least one model is not registered + # models are registered with admin in the 'admin.py' file - so we check + # if the app_config has an admin module before trying to laod it + if model_not_reg and hasattr(app_config.module, 'admin'): + reload(app_config.module.admin) + + def _get_plugin_path(self, plugin): + """Parse plugin path. + + The input can be eiter: + - a local file / dir + - a package + """ + try: + # for local path plugins + plugin_path = '.'.join(plugin.path().relative_to(settings.BASE_DIR).parts) + except ValueError: # pragma: no cover + # plugin is shipped as package - extract plugin module name + plugin_path = plugin.__module__.split('.')[0] + return plugin_path + + def _try_reload(self, cmd, *args, **kwargs): + """Wrapper to try reloading the apps. + + Throws an custom error that gets handled by the loading function. + """ + try: + cmd(*args, **kwargs) + return True, [] + except Exception as error: # pragma: no cover + handle_error(error) + + def _reload_apps(self, force_reload: bool = False, full_reload: bool = False): + """Internal: reload apps using django internal functions. + + Args: + force_reload (bool, optional): Also reload base apps. Defaults to False. + full_reload (bool, optional): Reload everything - including plugin mechanism. Defaults to False. + """ + # If full_reloading is set to true we do not want to set the flag + if not full_reload: + self.is_loading = True # set flag to disable loop reloading + if force_reload: + # we can not use the built in functions as we need to brute force the registry + apps.app_configs = OrderedDict() + apps.apps_ready = apps.models_ready = apps.loading = apps.ready = False + apps.clear_cache() + self._try_reload(apps.populate, settings.INSTALLED_APPS) + else: + self._try_reload(apps.set_installed_apps, settings.INSTALLED_APPS) + self.is_loading = False + # endregion + + @property + def has_app(self): + """This plugin is always an app with this plugin.""" + return True diff --git a/InvenTree/plugin/base/integration/ScheduleMixin.py b/InvenTree/plugin/base/integration/ScheduleMixin.py new file mode 100644 index 0000000000..6bbfe936ed --- /dev/null +++ b/InvenTree/plugin/base/integration/ScheduleMixin.py @@ -0,0 +1,211 @@ +"""Plugin mixin class for ScheduleMixin.""" +import logging + +from django.conf import settings +from django.db.utils import OperationalError, ProgrammingError + +from plugin.helpers import MixinImplementationError + +logger = logging.getLogger('inventree') + + +class ScheduleMixin: + """Mixin that provides support for scheduled tasks. + + Implementing classes must provide a dict object called SCHEDULED_TASKS, + which provides information on the tasks to be scheduled. + + SCHEDULED_TASKS = { + # Name of the task (will be prepended with the plugin name) + 'test_server': { + 'func': 'myplugin.tasks.test_server', # Python function to call (no arguments!) + 'schedule': "I", # Schedule type (see django_q.Schedule) + 'minutes': 30, # Number of minutes (only if schedule type = Minutes) + 'repeats': 5, # Number of repeats (leave blank for 'forever') + }, + 'member_func': { + 'func': 'my_class_func', # Note, without the 'dot' notation, it will call a class member function + 'schedule': "H", # Once per hour + }, + } + + Note: 'schedule' parameter must be one of ['I', 'H', 'D', 'W', 'M', 'Q', 'Y'] + + Note: The 'func' argument can take two different forms: + - Dotted notation e.g. 'module.submodule.func' - calls a global function with the defined path + - Member notation e.g. 'my_func' (no dots!) - calls a member function of the calling class + """ + + ALLOWABLE_SCHEDULE_TYPES = ['I', 'H', 'D', 'W', 'M', 'Q', 'Y'] + + # Override this in subclass model + SCHEDULED_TASKS = {} + + class MixinMeta: + """Meta options for this mixin.""" + + MIXIN_NAME = 'Schedule' + + def __init__(self): + """Register mixin.""" + super().__init__() + self.scheduled_tasks = self.get_scheduled_tasks() + self.validate_scheduled_tasks() + + self.add_mixin('schedule', 'has_scheduled_tasks', __class__) + + def _activate_mixin(self, plugins, *args, **kwargs): + """Activate scheudles from plugins with the ScheduleMixin.""" + logger.info('Activating plugin tasks') + + from common.models import InvenTreeSetting + + # List of tasks we have activated + task_keys = [] + + if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_SCHEDULE'): + + for _key, plugin in plugins: + + if plugin.mixin_enabled('schedule'): + + if plugin.is_active(): + # Only active tasks for plugins which are enabled + plugin.register_tasks() + task_keys += plugin.get_task_names() + + if len(task_keys) > 0: + logger.info(f"Activated {len(task_keys)} scheduled tasks") + + # Remove any scheduled tasks which do not match + # This stops 'old' plugin tasks from accumulating + try: + from django_q.models import Schedule + + scheduled_plugin_tasks = Schedule.objects.filter(name__istartswith="plugin.") + + deleted_count = 0 + + for task in scheduled_plugin_tasks: + if task.name not in task_keys: + task.delete() + deleted_count += 1 + + if deleted_count > 0: + logger.info(f"Removed {deleted_count} old scheduled tasks") # pragma: no cover + except (ProgrammingError, OperationalError): + # Database might not yet be ready + logger.warning("activate_integration_schedule failed, database not ready") + + def _deactivate_mixin(self): + """Deactivate ScheduleMixin. + + Currently nothing is done here. + """ + pass + + def get_scheduled_tasks(self): + """Returns `SCHEDULED_TASKS` context. + + Override if you want the scheduled tasks to be dynamic (influenced by settings for example). + """ + return getattr(self, 'SCHEDULED_TASKS', {}) + + @property + def has_scheduled_tasks(self): + """Are tasks defined for this plugin.""" + return bool(self.scheduled_tasks) + + def validate_scheduled_tasks(self): + """Check that the provided scheduled tasks are valid.""" + if not self.has_scheduled_tasks: + raise MixinImplementationError("SCHEDULED_TASKS not defined") + + for key, task in self.scheduled_tasks.items(): + + if 'func' not in task: + raise MixinImplementationError(f"Task '{key}' is missing 'func' parameter") + + if 'schedule' not in task: + raise MixinImplementationError(f"Task '{key}' is missing 'schedule' parameter") + + schedule = task['schedule'].upper().strip() + + if schedule not in self.ALLOWABLE_SCHEDULE_TYPES: + raise MixinImplementationError(f"Task '{key}': Schedule '{schedule}' is not a valid option") + + # If 'minutes' is selected, it must be provided! + if schedule == 'I' and 'minutes' not in task: + raise MixinImplementationError(f"Task '{key}' is missing 'minutes' parameter") + + def get_task_name(self, key): + """Task name for key.""" + # Generate a 'unique' task name + slug = self.plugin_slug() + return f"plugin.{slug}.{key}" + + def get_task_names(self): + """All defined task names.""" + # Returns a list of all task names associated with this plugin instance + return [self.get_task_name(key) for key in self.scheduled_tasks.keys()] + + def register_tasks(self): + """Register the tasks with the database.""" + try: + from django_q.models import Schedule + + for key, task in self.scheduled_tasks.items(): + + task_name = self.get_task_name(key) + + obj = { + 'name': task_name, + 'schedule_type': task['schedule'], + 'minutes': task.get('minutes', None), + 'repeats': task.get('repeats', -1), + } + + func_name = task['func'].strip() + + if '.' in func_name: + """Dotted notation indicates that we wish to run a globally defined function, from a specified Python module.""" + obj['func'] = func_name + else: + """Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin. This is managed by the plugin registry itself.""" + slug = self.plugin_slug() + obj['func'] = 'plugin.registry.call_plugin_function' + obj['args'] = f"'{slug}', '{func_name}'" + + if Schedule.objects.filter(name=task_name).exists(): + # Scheduled task already exists - update it! + logger.info(f"Updating scheduled task '{task_name}'") + instance = Schedule.objects.get(name=task_name) + for item in obj: + setattr(instance, item, obj[item]) + instance.save() + else: + logger.info(f"Adding scheduled task '{task_name}'") + # Create a new scheduled task + Schedule.objects.create(**obj) + + except (ProgrammingError, OperationalError): # pragma: no cover + # Database might not yet be ready + logger.warning("register_tasks failed, database not ready") + + def unregister_tasks(self): + """Deregister the tasks with the database.""" + try: + from django_q.models import Schedule + + for key, _ in self.scheduled_tasks.items(): + + task_name = self.get_task_name(key) + + try: + scheduled_task = Schedule.objects.get(name=task_name) + scheduled_task.delete() + except Schedule.DoesNotExist: + pass + except (ProgrammingError, OperationalError): # pragma: no cover + # Database might not yet be ready + logger.warning("unregister_tasks failed, database not ready") diff --git a/InvenTree/plugin/base/integration/SettingsMixin.py b/InvenTree/plugin/base/integration/SettingsMixin.py new file mode 100644 index 0000000000..35eb9c4169 --- /dev/null +++ b/InvenTree/plugin/base/integration/SettingsMixin.py @@ -0,0 +1,73 @@ +"""Plugin mixin class for SettingsMixin.""" +import logging + +from django.db.utils import OperationalError, ProgrammingError + +logger = logging.getLogger('inventree') + + +class SettingsMixin: + """Mixin that enables global settings for the plugin.""" + + class MixinMeta: + """Meta for mixin.""" + MIXIN_NAME = 'Settings' + + def __init__(self): + """Register mixin.""" + super().__init__() + self.add_mixin('settings', 'has_settings', __class__) + self.settings = getattr(self, 'SETTINGS', {}) + + def _activate_mixin(self, plugins, *args, **kwargs): + """Activate plugin settings. + + Add all defined settings form the plugins to a unified dict in the registry. + This dict is referenced by the PluginSettings for settings definitions. + """ + logger.info('Activating plugin settings') + + self.mixins_settings = {} + + for slug, plugin in plugins: + if plugin.mixin_enabled('settings'): + plugin_setting = plugin.settings + self.mixins_settings[slug] = plugin_setting + + def _deactivate_mixin(self): + """Deactivate all plugin settings.""" + logger.info('Deactivating plugin settings') + # clear settings cache + self.mixins_settings = {} + + @property + def has_settings(self): + """Does this plugin use custom global settings.""" + return bool(self.settings) + + def get_setting(self, key, cache=False): + """Return the 'value' of the setting associated with this plugin. + + Arguments: + key: The 'name' of the setting value to be retrieved + cache: Whether to use RAM cached value (default = False) + """ + from plugin.models import PluginSetting + + return PluginSetting.get_setting(key, plugin=self, cache=cache) + + def set_setting(self, key, value, user=None): + """Set plugin setting value by key.""" + from plugin.models import PluginConfig, PluginSetting + + try: + plugin, _ = PluginConfig.objects.get_or_create(key=self.plugin_slug(), name=self.plugin_name()) + except (OperationalError, ProgrammingError): # pragma: no cover + plugin = None + + if not plugin: # pragma: no cover + # Cannot find associated plugin model, return + logger.error(f"Plugin configuration not found for plugin '{self.slug}'") + return + + PluginSetting.set_setting(key, value, user, plugin=plugin) diff --git a/InvenTree/plugin/base/integration/UrlsMixin.py b/InvenTree/plugin/base/integration/UrlsMixin.py new file mode 100644 index 0000000000..b4c91bc683 --- /dev/null +++ b/InvenTree/plugin/base/integration/UrlsMixin.py @@ -0,0 +1,71 @@ +"""Plugin mixin class for UrlsMixin.""" +import logging + +from django.conf import settings +from django.urls import include, re_path + +from plugin.urls import PLUGIN_BASE + +logger = logging.getLogger('inventree') + + +class UrlsMixin: + """Mixin that enables custom URLs for the plugin.""" + + class MixinMeta: + """Meta options for this mixin.""" + + MIXIN_NAME = 'URLs' + + def __init__(self): + """Register mixin.""" + super().__init__() + self.add_mixin('urls', 'has_urls', __class__) + self.urls = self.setup_urls() + + def _activate_mixin(self, plugins, force_reload=False, full_reload: bool = False): + """Activate UrlsMixin plugins - add custom urls . + + Args: + plugins (dict): List of IntegrationPlugins that should be installed + force_reload (bool, optional): Only reload base apps. Defaults to False. + full_reload (bool, optional): Reload everything - including plugin mechanism. Defaults to False. + """ + from common.models import InvenTreeSetting + if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_URL'): + logger.info('Registering UrlsMixin Plugin') + urls_changed = False + # check whether an activated plugin extends UrlsMixin + for _key, plugin in plugins: + if plugin.mixin_enabled('urls'): + urls_changed = True + # if apps were changed or force loading base apps -> reload + if urls_changed or force_reload or full_reload: + # update urls - must be last as models must be registered for creating admin routes + self._update_urls() + + def setup_urls(self): + """Setup url endpoints for this plugin.""" + return getattr(self, 'URLS', None) + + @property + def base_url(self): + """Base url for this plugin.""" + return f'{PLUGIN_BASE}/{self.slug}/' + + @property + def internal_name(self): + """Internal url pattern name.""" + return f'plugin:{self.slug}:' + + @property + def urlpatterns(self): + """Urlpatterns for this plugin.""" + if self.has_urls: + return re_path(f'^{self.slug}/', include((self.urls, self.slug)), name=self.slug) + return None + + @property + def has_urls(self): + """Does this plugin use custom urls.""" + return bool(self.urls) diff --git a/InvenTree/plugin/base/integration/mixins.py b/InvenTree/plugin/base/integration/mixins.py index d1e842d416..55adb3e4c1 100644 --- a/InvenTree/plugin/base/integration/mixins.py +++ b/InvenTree/plugin/base/integration/mixins.py @@ -2,293 +2,15 @@ import json as json_pkg import logging -from importlib import reload -from typing import OrderedDict - -from django.apps import apps -from django.conf import settings -from django.contrib import admin -from django.db.utils import OperationalError, ProgrammingError -from django.urls import include, re_path import requests -from plugin.helpers import (MixinImplementationError, MixinNotImplementedError, - handle_error, render_template, render_text) -from plugin.urls import PLUGIN_BASE +from plugin.helpers import (MixinNotImplementedError, render_template, + render_text) logger = logging.getLogger('inventree') -class SettingsMixin: - """Mixin that enables global settings for the plugin.""" - - class MixinMeta: - """Meta for mixin.""" - MIXIN_NAME = 'Settings' - - def __init__(self): - """Register mixin.""" - super().__init__() - self.add_mixin('settings', 'has_settings', __class__) - self.settings = getattr(self, 'SETTINGS', {}) - - def _activate_mixin(self, plugins, *args, **kwargs): - """Activate plugin settings. - - Add all defined settings form the plugins to a unified dict in the registry. - This dict is referenced by the PluginSettings for settings definitions. - """ - logger.info('Activating plugin settings') - - self.mixins_settings = {} - - for slug, plugin in plugins: - if plugin.mixin_enabled('settings'): - plugin_setting = plugin.settings - self.mixins_settings[slug] = plugin_setting - - def _deactivate_mixin(self): - """Deactivate all plugin settings.""" - logger.info('Deactivating plugin settings') - # clear settings cache - self.mixins_settings = {} - - @property - def has_settings(self): - """Does this plugin use custom global settings.""" - return bool(self.settings) - - def get_setting(self, key, cache=False): - """Return the 'value' of the setting associated with this plugin. - - Arguments: - key: The 'name' of the setting value to be retrieved - cache: Whether to use RAM cached value (default = False) - """ - from plugin.models import PluginSetting - - return PluginSetting.get_setting(key, plugin=self, cache=cache) - - def set_setting(self, key, value, user=None): - """Set plugin setting value by key.""" - from plugin.models import PluginConfig, PluginSetting - - try: - plugin, _ = PluginConfig.objects.get_or_create(key=self.plugin_slug(), name=self.plugin_name()) - except (OperationalError, ProgrammingError): # pragma: no cover - plugin = None - - if not plugin: # pragma: no cover - # Cannot find associated plugin model, return - logger.error(f"Plugin configuration not found for plugin '{self.slug}'") - return - - PluginSetting.set_setting(key, value, user, plugin=plugin) - - -class ScheduleMixin: - """Mixin that provides support for scheduled tasks. - - Implementing classes must provide a dict object called SCHEDULED_TASKS, - which provides information on the tasks to be scheduled. - - SCHEDULED_TASKS = { - # Name of the task (will be prepended with the plugin name) - 'test_server': { - 'func': 'myplugin.tasks.test_server', # Python function to call (no arguments!) - 'schedule': "I", # Schedule type (see django_q.Schedule) - 'minutes': 30, # Number of minutes (only if schedule type = Minutes) - 'repeats': 5, # Number of repeats (leave blank for 'forever') - }, - 'member_func': { - 'func': 'my_class_func', # Note, without the 'dot' notation, it will call a class member function - 'schedule': "H", # Once per hour - }, - } - - Note: 'schedule' parameter must be one of ['I', 'H', 'D', 'W', 'M', 'Q', 'Y'] - - Note: The 'func' argument can take two different forms: - - Dotted notation e.g. 'module.submodule.func' - calls a global function with the defined path - - Member notation e.g. 'my_func' (no dots!) - calls a member function of the calling class - """ - - ALLOWABLE_SCHEDULE_TYPES = ['I', 'H', 'D', 'W', 'M', 'Q', 'Y'] - - # Override this in subclass model - SCHEDULED_TASKS = {} - - class MixinMeta: - """Meta options for this mixin.""" - - MIXIN_NAME = 'Schedule' - - def __init__(self): - """Register mixin.""" - super().__init__() - self.scheduled_tasks = self.get_scheduled_tasks() - self.validate_scheduled_tasks() - - self.add_mixin('schedule', 'has_scheduled_tasks', __class__) - - def _activate_mixin(self, plugins, *args, **kwargs): - """Activate scheudles from plugins with the ScheduleMixin.""" - logger.info('Activating plugin tasks') - - from common.models import InvenTreeSetting - - # List of tasks we have activated - task_keys = [] - - if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_SCHEDULE'): - - for _key, plugin in plugins: - - if plugin.mixin_enabled('schedule'): - - if plugin.is_active(): - # Only active tasks for plugins which are enabled - plugin.register_tasks() - task_keys += plugin.get_task_names() - - if len(task_keys) > 0: - logger.info(f"Activated {len(task_keys)} scheduled tasks") - - # Remove any scheduled tasks which do not match - # This stops 'old' plugin tasks from accumulating - try: - from django_q.models import Schedule - - scheduled_plugin_tasks = Schedule.objects.filter(name__istartswith="plugin.") - - deleted_count = 0 - - for task in scheduled_plugin_tasks: - if task.name not in task_keys: - task.delete() - deleted_count += 1 - - if deleted_count > 0: - logger.info(f"Removed {deleted_count} old scheduled tasks") # pragma: no cover - except (ProgrammingError, OperationalError): - # Database might not yet be ready - logger.warning("activate_integration_schedule failed, database not ready") - - def _deactivate_mixin(self): - """Deactivate ScheduleMixin. - - Currently nothing is done here. - """ - pass - - def get_scheduled_tasks(self): - """Returns `SCHEDULED_TASKS` context. - - Override if you want the scheduled tasks to be dynamic (influenced by settings for example). - """ - return getattr(self, 'SCHEDULED_TASKS', {}) - - @property - def has_scheduled_tasks(self): - """Are tasks defined for this plugin.""" - return bool(self.scheduled_tasks) - - def validate_scheduled_tasks(self): - """Check that the provided scheduled tasks are valid.""" - if not self.has_scheduled_tasks: - raise MixinImplementationError("SCHEDULED_TASKS not defined") - - for key, task in self.scheduled_tasks.items(): - - if 'func' not in task: - raise MixinImplementationError(f"Task '{key}' is missing 'func' parameter") - - if 'schedule' not in task: - raise MixinImplementationError(f"Task '{key}' is missing 'schedule' parameter") - - schedule = task['schedule'].upper().strip() - - if schedule not in self.ALLOWABLE_SCHEDULE_TYPES: - raise MixinImplementationError(f"Task '{key}': Schedule '{schedule}' is not a valid option") - - # If 'minutes' is selected, it must be provided! - if schedule == 'I' and 'minutes' not in task: - raise MixinImplementationError(f"Task '{key}' is missing 'minutes' parameter") - - def get_task_name(self, key): - """Task name for key.""" - # Generate a 'unique' task name - slug = self.plugin_slug() - return f"plugin.{slug}.{key}" - - def get_task_names(self): - """All defined task names.""" - # Returns a list of all task names associated with this plugin instance - return [self.get_task_name(key) for key in self.scheduled_tasks.keys()] - - def register_tasks(self): - """Register the tasks with the database.""" - try: - from django_q.models import Schedule - - for key, task in self.scheduled_tasks.items(): - - task_name = self.get_task_name(key) - - obj = { - 'name': task_name, - 'schedule_type': task['schedule'], - 'minutes': task.get('minutes', None), - 'repeats': task.get('repeats', -1), - } - - func_name = task['func'].strip() - - if '.' in func_name: - """Dotted notation indicates that we wish to run a globally defined function, from a specified Python module.""" - obj['func'] = func_name - else: - """Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin. This is managed by the plugin registry itself.""" - slug = self.plugin_slug() - obj['func'] = 'plugin.registry.call_plugin_function' - obj['args'] = f"'{slug}', '{func_name}'" - - if Schedule.objects.filter(name=task_name).exists(): - # Scheduled task already exists - update it! - logger.info(f"Updating scheduled task '{task_name}'") - instance = Schedule.objects.get(name=task_name) - for item in obj: - setattr(instance, item, obj[item]) - instance.save() - else: - logger.info(f"Adding scheduled task '{task_name}'") - # Create a new scheduled task - Schedule.objects.create(**obj) - - except (ProgrammingError, OperationalError): # pragma: no cover - # Database might not yet be ready - logger.warning("register_tasks failed, database not ready") - - def unregister_tasks(self): - """Deregister the tasks with the database.""" - try: - from django_q.models import Schedule - - for key, _ in self.scheduled_tasks.items(): - - task_name = self.get_task_name(key) - - try: - scheduled_task = Schedule.objects.get(name=task_name) - scheduled_task.delete() - except Schedule.DoesNotExist: - pass - except (ProgrammingError, OperationalError): # pragma: no cover - # Database might not yet be ready - logger.warning("unregister_tasks failed, database not ready") - - class ValidationMixin: """Mixin class that allows custom validation for various parts of InvenTree @@ -422,68 +144,6 @@ class ValidationMixin: return None -class UrlsMixin: - """Mixin that enables custom URLs for the plugin.""" - - class MixinMeta: - """Meta options for this mixin.""" - - MIXIN_NAME = 'URLs' - - def __init__(self): - """Register mixin.""" - super().__init__() - self.add_mixin('urls', 'has_urls', __class__) - self.urls = self.setup_urls() - - def _activate_mixin(self, plugins, force_reload=False, full_reload: bool = False): - """Activate UrlsMixin plugins - add custom urls . - - Args: - plugins (dict): List of IntegrationPlugins that should be installed - force_reload (bool, optional): Only reload base apps. Defaults to False. - full_reload (bool, optional): Reload everything - including plugin mechanism. Defaults to False. - """ - from common.models import InvenTreeSetting - if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_URL'): - logger.info('Registering UrlsMixin Plugin') - urls_changed = False - # check whether an activated plugin extends UrlsMixin - for _key, plugin in plugins: - if plugin.mixin_enabled('urls'): - urls_changed = True - # if apps were changed or force loading base apps -> reload - if urls_changed or force_reload or full_reload: - # update urls - must be last as models must be registered for creating admin routes - self._update_urls() - - def setup_urls(self): - """Setup url endpoints for this plugin.""" - return getattr(self, 'URLS', None) - - @property - def base_url(self): - """Base url for this plugin.""" - return f'{PLUGIN_BASE}/{self.slug}/' - - @property - def internal_name(self): - """Internal url pattern name.""" - return f'plugin:{self.slug}:' - - @property - def urlpatterns(self): - """Urlpatterns for this plugin.""" - if self.has_urls: - return re_path(f'^{self.slug}/', include((self.urls, self.slug)), name=self.slug) - return None - - @property - def has_urls(self): - """Does this plugin use custom urls.""" - return bool(self.urls) - - class NavigationMixin: """Mixin that enables custom navigation links with the plugin.""" @@ -530,186 +190,6 @@ class NavigationMixin: return getattr(self, 'NAVIGATION_TAB_ICON', "fas fa-question") -class AppMixin: - """Mixin that enables full django app functions for a plugin.""" - - class MixinMeta: - """Meta options for this mixin.""" - - MIXIN_NAME = 'App registration' - - def __init__(self): - """Register mixin.""" - super().__init__() - self.add_mixin('app', 'has_app', __class__) - - def _activate_mixin(self, plugins, force_reload=False, full_reload: bool = False): - """Activate AppMixin plugins - add custom apps and reload. - - Args: - plugins (dict): List of IntegrationPlugins that should be installed - force_reload (bool, optional): Only reload base apps. Defaults to False. - full_reload (bool, optional): Reload everything - including plugin mechanism. Defaults to False. - """ - from common.models import InvenTreeSetting - - if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_APP'): - logger.info('Registering IntegrationPlugin apps') - apps_changed = False - - # add them to the INSTALLED_APPS - for _key, plugin in plugins: - if plugin.mixin_enabled('app'): - plugin_path = self._get_plugin_path(plugin) - if plugin_path not in settings.INSTALLED_APPS: - settings.INSTALLED_APPS += [plugin_path] - self.installed_apps += [plugin_path] - apps_changed = True - # if apps were changed or force loading base apps -> reload - if apps_changed or force_reload: - # first startup or force loading of base apps -> registry is prob false - if self.apps_loading or force_reload: - self.apps_loading = False - self._reload_apps(force_reload=True, full_reload=full_reload) - else: - self._reload_apps(full_reload=full_reload) - - # rediscover models/ admin sites - self._reregister_contrib_apps() - - # update urls - must be last as models must be registered for creating admin routes - self._update_urls() - - def _deactivate_mixin(self): - """Deactivate AppMixin plugins - some magic required.""" - # unregister models from admin - for plugin_path in self.installed_apps: - models = [] # the modelrefs need to be collected as poping an item in a iter is not welcomed - app_name = plugin_path.split('.')[-1] - try: - app_config = apps.get_app_config(app_name) - - # check all models - for model in app_config.get_models(): - # remove model from admin site - try: - admin.site.unregister(model) - except Exception: # pragma: no cover - pass - models += [model._meta.model_name] - except LookupError: # pragma: no cover - # if an error occurs the app was never loaded right -> so nothing to do anymore - logger.debug(f'{app_name} App was not found during deregistering') - break - - # unregister the models (yes, models are just kept in multilevel dicts) - for model in models: - # remove model from general registry - apps.all_models[plugin_path].pop(model) - - # clear the registry for that app - # so that the import trick will work on reloading the same plugin - # -> the registry is kept for the whole lifecycle - if models and app_name in apps.all_models: - apps.all_models.pop(app_name) - - # remove plugin from installed_apps - self._clean_installed_apps() - - # reset load flag and reload apps - settings.INTEGRATION_APPS_LOADED = False - self._reload_apps() - - # update urls to remove the apps from the site admin - self._update_urls() - - # region helpers - def _reregister_contrib_apps(self): - """Fix reloading of contrib apps - models and admin. - - This is needed if plugins were loaded earlier and then reloaded as models and admins rely on imports. - Those register models and admin in their respective objects (e.g. admin.site for admin). - """ - for plugin_path in self.installed_apps: - try: - app_name = plugin_path.split('.')[-1] - app_config = apps.get_app_config(app_name) - except LookupError: # pragma: no cover - # the plugin was never loaded correctly - logger.debug(f'{app_name} App was not found during deregistering') - break - - # reload models if they were set - # models_module gets set if models were defined - even after multiple loads - # on a reload the models registery is empty but models_module is not - if app_config.models_module and len(app_config.models) == 0: - reload(app_config.models_module) - - # check for all models if they are registered with the site admin - model_not_reg = False - for model in app_config.get_models(): - if not admin.site.is_registered(model): - model_not_reg = True - - # reload admin if at least one model is not registered - # models are registered with admin in the 'admin.py' file - so we check - # if the app_config has an admin module before trying to laod it - if model_not_reg and hasattr(app_config.module, 'admin'): - reload(app_config.module.admin) - - def _get_plugin_path(self, plugin): - """Parse plugin path. - - The input can be eiter: - - a local file / dir - - a package - """ - try: - # for local path plugins - plugin_path = '.'.join(plugin.path().relative_to(settings.BASE_DIR).parts) - except ValueError: # pragma: no cover - # plugin is shipped as package - extract plugin module name - plugin_path = plugin.__module__.split('.')[0] - return plugin_path - - def _try_reload(self, cmd, *args, **kwargs): - """Wrapper to try reloading the apps. - - Throws an custom error that gets handled by the loading function. - """ - try: - cmd(*args, **kwargs) - return True, [] - except Exception as error: # pragma: no cover - handle_error(error) - - def _reload_apps(self, force_reload: bool = False, full_reload: bool = False): - """Internal: reload apps using django internal functions. - - Args: - force_reload (bool, optional): Also reload base apps. Defaults to False. - full_reload (bool, optional): Reload everything - including plugin mechanism. Defaults to False. - """ - # If full_reloading is set to true we do not want to set the flag - if not full_reload: - self.is_loading = True # set flag to disable loop reloading - if force_reload: - # we can not use the built in functions as we need to brute force the registry - apps.app_configs = OrderedDict() - apps.apps_ready = apps.models_ready = apps.loading = apps.ready = False - apps.clear_cache() - self._try_reload(apps.populate, settings.INSTALLED_APPS) - else: - self._try_reload(apps.set_installed_apps, settings.INSTALLED_APPS) - self.is_loading = False - # endregion - - @property - def has_app(self): - """This plugin is always an app with this plugin.""" - return True - - class APICallMixin: """Mixin that enables easier API calls for a plugin. diff --git a/InvenTree/plugin/mixins/__init__.py b/InvenTree/plugin/mixins/__init__.py index c42c57b528..7fba7d1110 100644 --- a/InvenTree/plugin/mixins/__init__.py +++ b/InvenTree/plugin/mixins/__init__.py @@ -6,10 +6,13 @@ from common.notifications import (BulkNotificationMethod, from ..base.action.mixins import ActionMixin from ..base.barcodes.mixins import BarcodeMixin from ..base.event.mixins import EventMixin -from ..base.integration.mixins import (APICallMixin, AppMixin, NavigationMixin, - PanelMixin, ScheduleMixin, - SettingsContentMixin, SettingsMixin, - UrlsMixin, ValidationMixin) +from ..base.integration.AppMixin import AppMixin +from ..base.integration.mixins import (APICallMixin, NavigationMixin, + PanelMixin, SettingsContentMixin, + ValidationMixin) +from ..base.integration.ScheduleMixin import ScheduleMixin +from ..base.integration.SettingsMixin import SettingsMixin +from ..base.integration.UrlsMixin import UrlsMixin from ..base.label.mixins import LabelPrintingMixin from ..base.locate.mixins import LocateMixin diff --git a/InvenTree/plugin/registry.py b/InvenTree/plugin/registry.py index 2f65f881bf..3ade312ff0 100644 --- a/InvenTree/plugin/registry.py +++ b/InvenTree/plugin/registry.py @@ -34,8 +34,10 @@ logger = logging.getLogger('inventree') class PluginsRegistry: """The PluginsRegistry class.""" - from .base.integration.mixins import (AppMixin, ScheduleMixin, - SettingsMixin, UrlsMixin) + from .base.integration.AppMixin import AppMixin + from .base.integration.ScheduleMixin import ScheduleMixin + from .base.integration.SettingsMixin import SettingsMixin + from .base.integration.UrlsMixin import UrlsMixin DEFAULT_MIXIN_ORDER = [SettingsMixin, ScheduleMixin, AppMixin, UrlsMixin] def __init__(self, mixin_order: list = None) -> None: