move activation/deactivation to mixins

InvenTree/plugin/base/integration/mixins.py

@@ -2,14 +2,19 @@
 
 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,
-                            render_template, render_text)
+                            handle_error, render_template, render_text)
 from plugin.urls import PLUGIN_BASE
 
 logger = logging.getLogger('inventree')
@@ -28,6 +33,27 @@ class SettingsMixin:
         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."""
@@ -106,6 +132,56 @@ class ScheduleMixin:
 
         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.
 
@@ -360,6 +436,27 @@ class UrlsMixin:
         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)
@@ -446,6 +543,167 @@ class AppMixin:
         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."""

InvenTree/plugin/registry.py

@@ -9,11 +9,9 @@ import importlib
 import logging
 import os
 import subprocess
-from importlib import reload
 from pathlib import Path
-from typing import Dict, List, OrderedDict
+from typing import Dict, List
 
-from django.apps import apps
 from django.conf import settings
 from django.contrib import admin
 from django.db.utils import IntegrityError, OperationalError, ProgrammingError
@@ -469,239 +467,17 @@ class PluginsRegistry:
         plugins = self.plugins.items()
         logger.info(f'Found {len(plugins)} active plugins')
 
-        self.activate_plugin_settings(plugins)
+        for mixin in self.mixin_order:
-        self.activate_plugin_schedule(plugins)
+            mixin._activate_mixin(mixin, plugins, force_reload=force_reload, full_reload=full_reload)
-        self.activate_plugin_app(plugins, force_reload=force_reload, full_reload=full_reload)
-        self.activate_plugin_url(plugins, force_reload=force_reload, full_reload=full_reload)
 
     def _deactivate_plugins(self):
         """Run deactivation functions for all plugins."""
-        self.deactivate_plugin_app()
+
-        self.deactivate_plugin_schedule()
+        for mixin in self.mixin_order:
-        self.deactivate_plugin_settings()
+            mixin._deactivate_mixin()
     # endregion
 
     # region mixin specific loading ...
-    def activate_plugin_settings(self, plugins):
-        """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_plugin_settings(self):
-        """Deactivate all plugin settings."""
-        logger.info('Deactivating plugin settings')
-        # clear settings cache
-        self.mixins_settings = {}
-
-    def activate_plugin_schedule(self, plugins):
-        """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_plugin_schedule(self):
-        """Deactivate ScheduleMixin.
-
-        Currently nothing is done here.
-        """
-        pass
-
-    def activate_plugin_app(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 activate_plugin_url(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 _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 deactivate_plugin_app(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()
-
     def _clean_installed_apps(self):
         for plugin in self.installed_apps:
             if plugin in settings.INSTALLED_APPS:
@@ -730,37 +506,6 @@ class PluginsRegistry:
         # Replace frontendpatterns
         global_pattern[0] = re_path('', include(urlpattern))
         clear_url_caches()
-
-    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
-
-    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)
     # endregion