Mirrored_Repos/InvenTree
1
0
Fork 0
mirror of https://github.com/inventree/InvenTree synced 2024-08-30 18:33:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

move loaded/unloaded mixins out into seperate modules

Browse Source
This commit is contained in:
Matthias Mair 2023-02-13 21:19:51 +01:00
parent 44609c5b27
commit de59bbf2be
No known key found for this signature in database
GPG Key ID: A593429DDA23B66A
7 changed files with 560 additions and 528 deletions

192
InvenTree/plugin/base/integration/AppMixin.py Normal file
View File

@ -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

211
InvenTree/plugin/base/integration/ScheduleMixin.py Normal file
View File

@ -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")

73
InvenTree/plugin/base/integration/SettingsMixin.py Normal file
View File

@ -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)

71
InvenTree/plugin/base/integration/UrlsMixin.py Normal file
View File

@ -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)

524
InvenTree/plugin/base/integration/mixins.py
View File

@ -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.

11
InvenTree/plugin/mixins/__init__.py
View File

@ -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

6
InvenTree/plugin/registry.py
View File

@ -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: