Coverage for src/mafw/plugin_manager.py: 100%

1# Copyright 2025–2026 European Union 

2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu) 

3# SPDX-License-Identifier: EUPL-1.2 

4""" 

5Plugin management system for MAFw framework. 

6 

7This module provides the core functionality for loading and managing plugins within the MAFw framework. 

8It supports loading various types of plugins including processors, standard tables, database models, and 

9user interfaces. 

10 

11The plugin manager uses the pluggy library to handle plugin discovery and registration 

12through entry points and hooks. 

13 

14When plugins are loaded using the :meth:`.MAFwPluginManager.load_plugins` function, the job is divided into multiple 

15threads to improve performance. 

16 

17Key features: 

18 - Asynchronous plugin loading with progress indication 

19 - Support for both internal and external plugins 

20 - Type-safe plugin handling with proper data structures 

21 - Logging integration for monitoring plugin loading processes 

22 - Global plugin manager singleton for consistent access 

23 

24The module defines several key components: 

25 - :class:`.LoadedPlugins`: Data container for loaded plugins 

26 - :class:`.MAFwPluginManager`: Main plugin manager class 

27 - :func:`.get_plugin_manager`: Factory function for accessing the plugin manager 

28 

29Plugin types supported: 

30 - Processors (`processors`): Classes that implement data processing logic 

31 - Database Modules (`db_modules`): Model modules for database interaction 

32 - User Interfaces (`ui`): UI implementations for different interfaces 

33 

34.. versionchanged:: v2.0.0 

35 Complete refactoring of the plugin manager system. 

36 

37""" 

38 

39import importlib 

40import itertools 

41import logging 

42import time 

43from collections.abc import Iterable 

44from concurrent.futures import Future, ThreadPoolExecutor 

45from dataclasses import dataclass, field 

46from typing import TYPE_CHECKING, Any, Literal, cast 

47 

48import pluggy 

49 

50from mafw import hookspecs, plugins 

51from mafw.lazy_import import ProcessorClassProtocol, UserInterfaceClassProtocol 

52 

53if TYPE_CHECKING: 

54 pass 

55 

56log = logging.getLogger(__name__) 

57 

58 

59@dataclass 

60class LoadedPlugins: 

61 """ 

62 Container class for storing loaded plugins of various types. 

63 

64 This dataclass holds collections of different plugin types that have been loaded 

65 by the :class:`MAFwPluginManager`. It provides organized storage for processors, 

66 database modules, and user interfaces. 

67 

68 .. versionadded:: v2.0.0 

69 """ 

70 

71 processor_list: list[ProcessorClassProtocol] = field(default_factory=list) 

72 # List[Type['Processor'] | 'LazyPlugin'] = field(default_factory=list) 

73 """List of loaded processor classes.""" 

74 

75 processor_dict: dict[str, ProcessorClassProtocol] = field(default_factory=dict) 

76 # Dict[str, Type['Processor'] | 'LazyPlugin'] = field(default_factory=dict) 

77 """Dictionary mapping processor names to their classes.""" 

78 

79 db_model_modules: list[str] = field(default_factory=list) 

80 """List of database model module names.""" 

81 

82 ui_list: list[UserInterfaceClassProtocol] = field(default_factory=list) 

83 # List[Type['UserInterfaceBase'] | 'LazyPlugin'] = field(default_factory=list) 

84 """List of loaded user interface classes.""" 

85 

86 ui_dict: dict[str, UserInterfaceClassProtocol] = field(default_factory=dict) 

87 # Dict[str, Type['UserInterfaceBase'] | 'LazyPlugin'] = field(default_factory=dict) 

88 """Dictionary mapping user interface names to their classes.""" 

89 

90 

91PluginTypes = Literal['processors', 'db_modules', 'ui'] 

92"""Type alias for accepted types of plugins.""" 

93 

94 

95def _as_processor_result(obj: object) -> tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]]: 

96 """ 

97 Cast an object to the expected processor result type. 

98 

99 This helper function is used to convert the raw result from plugin loading 

100 operations into the expected tuple format for processors. 

101 

102 .. versionadded:: v2.0.0 

103 

104 :param obj: The object to cast 

105 :type obj: object 

106 :return: A tuple containing a list of processor classes and a dictionary mapping 

107 processor names to their classes 

108 :rtype: tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]] 

109 """ 

110 return cast(tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]], obj) 

111 

112 

113def _as_ui_result(obj: object) -> tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]]: 

114 """ 

115 Cast an object to the expected UI result type. 

116 

117 This helper function is used to convert the raw result from plugin loading 

118 operations into the expected tuple format for user interfaces. 

119 

120 .. versionadded:: v2.0.0 

121 

122 :param obj: The object to cast 

123 :type obj: object 

124 :return: A tuple containing a list of UI classes and a dictionary mapping 

125 UI names to their classes 

126 :rtype: tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]] 

127 """ 

128 return cast(tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]], obj) 

129 

130 

131def _as_db_module_result(obj: object) -> list[str]: 

132 """ 

133 Cast an object to the expected database module result type. 

134 

135 This helper function is used to convert the raw result from plugin loading 

136 operations into the expected list format for database modules. 

137 

138 .. versionadded:: v2.0.0 

139 

140 :param obj: The object to cast 

141 :type obj: object 

142 :return: A list of database module names 

143 :rtype: list[str] 

144 """ 

145 return cast(list[str], obj) 

146 

147 

148class MAFwPluginManager(pluggy.PluginManager): 

149 """ 

150 The MAFw plugin manager. 

151 

152 The MAFwPluginManager class manages the loading and registration of plugins within the MAFw framework. 

153 It supports asynchronous loading of various plugin types, including processors, database modules, 

154 and user interfaces, using a thread pool executor for improved performance. 

155 

156 The class provides methods to load each type of plugin and handles delayed status messages if loading takes 

157 longer than expected. 

158 

159 .. versionadded:: v2.0.0 

160 """ 

161 

162 max_loading_delay = 1 # sec 

163 """ 

164 Loading delay before displaying a message. 

165 

166 If the loading of the external plugins is taking more than this value, a message is displayed to inform the user. 

167 """ 

168 

169 def __init__(self, project_name: str = 'mafw'): 

170 super().__init__(project_name) 

171 self._executor = ThreadPoolExecutor(max_workers=4) 

172 

173 def load_db_models_plugins(self) -> list[str]: 

174 """ 

175 Load database model modules from the plugin manager. 

176 

177 This method retrieves all database model modules registered through the plugin manager's 

178 :meth:`~mafw.plugins.register_db_model_modules` hook and imports them. 

179 

180 :returns: List of database model module names 

181 :rtype: list[str] 

182 """ 

183 log.debug('Starting database model plugins...') 

184 db_model_module_list = list(itertools.chain(*self.hook.register_db_model_modules())) 

185 for module in db_model_module_list: 

186 importlib.import_module(module) 

187 log.debug('Finished database model plugins') 

188 return db_model_module_list 

189 

190 def load_processor_plugins(self) -> tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]]: 

191 """ 

192 Load available processor plugins from the plugin manager. 

193 

194 This method retrieves all processor plugins registered through the plugin manager's 

195 :meth:`~mafw.plugins.register_processors` hook. 

196 :meth:`~mafw.plugins.register_processors` hook. 

197 

198 :returns: A tuple containing: 

199 - List of available processor classes 

200 - Dictionary mapping processor names to their classes 

201 :rtype: tuple[list[type[processor.Processor]], dict[str, type[processor.Processor]]] 

202 """ 

203 log.debug('Starting processor plugins...') 

204 lst = list(itertools.chain(*self.hook.register_processors())) 

205 dct = {} 

206 for p in lst: 

207 if hasattr(p, 'plugin_name'): # LazyPlugin case 

208 key = p.plugin_name 

209 else: 

210 key = p.__name__ 

211 dct[key] = p 

212 log.debug('Finished processor plugins') 

213 return lst, dct 

214 

215 def load_user_interface_plugins( 

216 self, 

217 ) -> tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]]: 

218 """ 

219 Load available user interface plugins from the plugin manager. 

220 

221 This method retrieves all user interface plugins registered through the plugin manager's 

222 :meth:`~mafw.plugins.register_user_interfaces` hook. 

223 

224 :returns: A tuple containing: 

225 - List of available user interface classes 

226 - Dictionary mapping user interface names to their classes 

227 :rtype: tuple[list[type[UserInterfaceBase]], dict[str, type[UserInterfaceBase]]] 

228 """ 

229 log.debug('Start loading user interface plugins...') 

230 lst = list(itertools.chain(*self.hook.register_user_interfaces())) 

231 dct = {ui.name: ui for ui in lst} 

232 log.debug('Finished loading user interface plugins') 

233 return lst, dct 

234 

235 def _delayed_status_message(self, futures: list[Future[Any]]) -> None: 

236 """ 

237 Display a warning message if plugin loading takes longer than expected. 

238 

239 This method is called after a delay to check if all plugin loading operations 

240 have completed. If not, it logs a warning message to inform the user that 

241 plugin loading is taking longer than expected. 

242 

243 :param futures: List of futures representing ongoing plugin loading operations 

244 :type futures: list[concurrent.futures.Future] 

245 """ 

246 time.sleep(self.max_loading_delay) 

247 if not all(f.done() for f in futures): 

248 log.warning('Plugin loading is taking longer than expected, please be patient.') 

249 

250 def load_plugins(self, plugins_to_load: Iterable[PluginTypes]) -> LoadedPlugins: 

251 """ 

252 Load plugins of specified types in multiple threads. 

253 

254 This method loads plugins of the specified types using a thread pool executor 

255 for improved performance. It handles different plugin types including processors, 

256 standard tables, database modules, and user interfaces. 

257 

258 :param plugins_to_load: Iterable of plugin types to load 

259 :type plugins_to_load: Iterable[:obj:`PluginTypes`] 

260 :return: Container with loaded plugins of all requested types 

261 :rtype: :obj:`LoadedPlugins` 

262 """ 

263 plugins_to_load = list(dict.fromkeys(plugins_to_load)) 

264 if not plugins_to_load: 

265 return LoadedPlugins() 

266 

267 lut = { 

268 'processors': self.load_processor_plugins, 

269 'db_modules': self.load_db_models_plugins, 

270 'ui': self.load_user_interface_plugins, 

271 } 

272 

273 # drop invalid plugin types 

274 plugins_to_load = [p for p in plugins_to_load if p in lut] 

275 if not plugins_to_load: 

276 return LoadedPlugins() 

277 

278 log.debug(f'Status message will appear if loading takes > {self.max_loading_delay}s') 

279 

280 # Submit tasks to the executor 

281 futures: list[Future[Any]] = [] 

282 for plugin_type in plugins_to_load: 

283 fut = self._executor.submit(lut[plugin_type]) 

284 futures.append(fut) 

285 

286 # Start delayed status thread 

287 status_thread = self._executor.submit(self._delayed_status_message, futures) 

288 

289 # Wait for completion 

290 results = [] 

291 try: 

292 for fut in futures: 

293 results.append(fut.result()) # will re-raise exceptions 

294 finally: 

295 # When all tasks finished, no need for the warning message 

296 if not status_thread.done(): 

297 status_thread.cancel() 

298 

299 # Assemble output 

300 plugins_ = LoadedPlugins() 

301 idx = 0 

302 

303 for plugin_type in plugins_to_load: 

304 result = results[idx] 

305 

306 if plugin_type == 'processors': 

307 plugins_.processor_list, plugins_.processor_dict = _as_processor_result(result) 

308 

309 elif plugin_type == 'ui': 

310 plugins_.ui_list, plugins_.ui_dict = _as_ui_result(result) 

311 

312 else: # if plugin_type == 'db_modules': 

313 plugins_.db_model_modules = _as_db_module_result(result) 

314 

315 idx += 1 

316 

317 return plugins_ 

318 

319 

320global_mafw_plugin_manager: dict[str, 'MAFwPluginManager'] = {} 

321"""The global mafw plugin manager dictionary.""" 

322 

323 

324def get_plugin_manager(force_recreate: bool = False) -> 'MAFwPluginManager': 

325 """ 

326 Create a new or return an existing plugin manager for a given project 

327 

328 :param force_recreate: Flag to force the creation of a new plugin manager. Defaults to False 

329 :type force_recreate: bool, Optional 

330 :return: The plugin manager 

331 :rtype: pluggy.PluginManager 

332 """ 

333 if 'mafw' in global_mafw_plugin_manager and force_recreate: 

334 del global_mafw_plugin_manager['mafw'] 

335 

336 if 'mafw' not in global_mafw_plugin_manager: 

337 pm = MAFwPluginManager('mafw') 

338 pm.add_hookspecs(hookspecs) 

339 pm.load_setuptools_entrypoints('mafw') 

340 pm.register(plugins) 

341 global_mafw_plugin_manager['mafw'] = pm 

342 

343 return global_mafw_plugin_manager['mafw']