Coverage for src/backend/InvenTree/plugin/base/integration/ScheduleMixin.py: 26%

1"""Plugin mixin class for ScheduleMixin.""" 

2 

3from django.conf import settings 

4from django.db.utils import OperationalError, ProgrammingError 

5 

6import structlog 

7 

8from common.settings import get_global_setting 

9from plugin import PluginMixinEnum 

10from plugin.helpers import MixinImplementationError 

11 

12logger = structlog.get_logger('inventree') 

13 

14 

15class ScheduleMixin: 

16 """Mixin that provides support for scheduled tasks. 

17 

18 Implementing classes must provide a dict object called SCHEDULED_TASKS, 

19 which provides information on the tasks to be scheduled. 

20 

21 SCHEDULED_TASKS = { 

22 # Name of the task (will be prepended with the plugin name) 

23 'test_server': { 

24 'func': 'myplugin.tasks.test_server', # Python function to call (no arguments!) 

25 'schedule': "I", # Schedule type (see django_q.Schedule) 

26 'minutes': 30, # Number of minutes (only if schedule type = Minutes) 

27 'repeats': 5, # Number of repeats (leave blank for 'forever') 

28 }, 

29 'member_func': { 

30 'func': 'my_class_func', # Note, without the 'dot' notation, it will call a class member function 

31 'schedule': "H", # Once per hour 

32 }, 

33 } 

34 

35 Note: 'schedule' parameter must be one of ['I', 'H', 'D', 'W', 'M', 'Q', 'Y'] 

36 

37 Note: The 'func' argument can take two different forms: 

38 - Dotted notation e.g. 'module.submodule.func' - calls a global function with the defined path 

39 - Member notation e.g. 'my_func' (no dots!) - calls a member function of the calling class 

40 """ 

41 

42 ALLOWABLE_SCHEDULE_TYPES = ['I', 'H', 'D', 'W', 'M', 'Q', 'Y'] 

43 

44 # Override this in subclass model 

45 SCHEDULED_TASKS = {} 

46 

47 class MixinMeta: 

48 """Meta options for this mixin.""" 

49 

50 MIXIN_NAME = 'Schedule' 

51 

52 def __init__(self): 

53 """Register mixin.""" 

54 super().__init__() 

55 

56 self.scheduled_tasks = [] 

57 

58 self.add_mixin(PluginMixinEnum.SCHEDULE, 'has_scheduled_tasks', __class__) 

59 

60 @classmethod 

61 def _activate_mixin(cls, registry, plugins, *args, **kwargs): 

62 """Activate schedules from plugins with the ScheduleMixin.""" 

63 logger.debug('Activating plugin tasks') 

64 

65 # List of tasks we have activated 

66 task_keys = [] 

67 

68 if settings.PLUGIN_TESTING or get_global_setting('ENABLE_PLUGINS_SCHEDULE'): 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 for _key, plugin in plugins: 

70 if ( 

71 plugin 

72 and plugin.is_active() 

73 and plugin.mixin_enabled(PluginMixinEnum.SCHEDULE) 

74 ): 

75 # Only active tasks for plugins which are enabled 

76 plugin.register_tasks() 

77 task_keys += plugin.get_task_names() 

78 

79 if len(task_keys) > 0: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 logger.info('Activated %s scheduled tasks', len(task_keys)) 

81 

82 # Remove any scheduled tasks which do not match 

83 # This stops 'old' plugin tasks from accumulating 

84 try: 

85 from django_q.models import Schedule 

86 

87 scheduled_plugin_tasks = Schedule.objects.filter( 

88 name__istartswith='plugin.' 

89 ) 

90 

91 deleted_count = 0 

92 

93 for task in scheduled_plugin_tasks: 93 ↛ 94line 93 didn't jump to line 94 because the loop on line 93 never started

94 if task.name not in task_keys: 

95 task.delete() 

96 deleted_count += 1 

97 

98 if deleted_count > 0: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 logger.info( 

100 'Removed %s old scheduled tasks', deleted_count 

101 ) # pragma: no cover 

102 except (ProgrammingError, OperationalError): 

103 # Database might not yet be ready 

104 logger.warning('activate_integration_schedule failed, database not ready') 

105 

106 def get_scheduled_tasks(self): 

107 """Returns `SCHEDULED_TASKS` context. 

108 

109 Override if you want the scheduled tasks to be dynamic (influenced by settings for example). 

110 """ 

111 return getattr(self, 'SCHEDULED_TASKS', {}) 

112 

113 @property 

114 def has_scheduled_tasks(self): 

115 """Are tasks defined for this plugin.""" 

116 return bool(self.get_scheduled_tasks()) 

117 

118 def validate_scheduled_tasks(self): 

119 """Check that the provided scheduled tasks are valid.""" 

120 if not self.has_scheduled_tasks: 

121 raise MixinImplementationError('SCHEDULED_TASKS not defined') 

122 

123 for key, task in self.scheduled_tasks.items(): 

124 if 'func' not in task: 

125 raise MixinImplementationError( 

126 f"Task '{key}' is missing 'func' parameter" 

127 ) 

128 

129 if 'schedule' not in task: 

130 raise MixinImplementationError( 

131 f"Task '{key}' is missing 'schedule' parameter" 

132 ) 

133 

134 schedule = task['schedule'].upper().strip() 

135 

136 if schedule not in self.ALLOWABLE_SCHEDULE_TYPES: 

137 raise MixinImplementationError( 

138 f"Task '{key}': Schedule '{schedule}' is not a valid option" 

139 ) 

140 

141 # If 'minutes' is selected, it must be provided! 

142 if schedule == 'I' and 'minutes' not in task: 

143 raise MixinImplementationError( 

144 f"Task '{key}' is missing 'minutes' parameter" 

145 ) 

146 

147 def get_task_name(self, key): 

148 """Task name for key.""" 

149 # Generate a 'unique' task name 

150 slug = self.plugin_slug() 

151 return f'plugin.{slug}.{key}' 

152 

153 def get_task_names(self): 

154 """All defined task names.""" 

155 # Returns a list of all task names associated with this plugin instance 

156 return [self.get_task_name(key) for key in self.scheduled_tasks] 

157 

158 def register_tasks(self): 

159 """Register the tasks with the database.""" 

160 self.scheduled_tasks = self.get_scheduled_tasks() 

161 self.validate_scheduled_tasks() 

162 

163 try: 

164 from django_q.models import Schedule 

165 

166 for key, task in self.scheduled_tasks.items(): 

167 task_name = self.get_task_name(key) 

168 

169 obj = { 

170 'name': task_name, 

171 'schedule_type': task['schedule'], 

172 'minutes': task.get('minutes', None), 

173 'repeats': task.get('repeats', -1), 

174 } 

175 

176 func_name = task['func'].strip() 

177 

178 if '.' in func_name: 

179 """Dotted notation indicates that we wish to run a globally defined function, from a specified Python module.""" 

180 obj['func'] = func_name 

181 else: 

182 """Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin. This is managed by the plugin registry itself.""" 

183 slug = self.plugin_slug() 

184 obj['func'] = 'plugin.registry.call_plugin_function' 

185 obj['args'] = f"'{slug}', '{func_name}'" 

186 

187 tasks = Schedule.objects.filter(name=task_name) 

188 

189 if len(tasks) > 1: 

190 logger.info( 

191 "Found multiple tasks; Adding a new scheduled task '%s'", 

192 task_name, 

193 ) 

194 tasks.delete() 

195 Schedule.objects.create(**obj) 

196 elif len(tasks) == 1: 

197 # Scheduled task already exists - update it! 

198 logger.info("Updating scheduled task '%s'", task_name) 

199 

200 if instance := tasks.first(): 

201 for item in obj: 

202 setattr(instance, item, obj[item]) 

203 instance.save() 

204 else: 

205 logger.info("Adding scheduled task '%s'", task_name) 

206 # Create a new scheduled task 

207 Schedule.objects.create(**obj) 

208 

209 except (ProgrammingError, OperationalError): # pragma: no cover 

210 # Database might not yet be ready 

211 logger.warning('register_tasks failed, database not ready') 

212 

213 def unregister_tasks(self): 

214 """Deregister the tasks with the database.""" 

215 try: 

216 from django_q.models import Schedule 

217 

218 for key in self.scheduled_tasks: 

219 task_name = self.get_task_name(key) 

220 

221 try: 

222 scheduled_task = Schedule.objects.get(name=task_name) 

223 scheduled_task.delete() 

224 except Schedule.DoesNotExist: 

225 pass 

226 except (ProgrammingError, OperationalError): # pragma: no cover 

227 # Database might not yet be ready 

228 logger.warning('unregister_tasks failed, database not ready')