1""" 2 lml.plugin 3 ~~~~~~~~~~~~~~~~~~~ 4 5 lml divides the plugins into two category: load-me-later plugins and 6 load-me-now ones. load-me-later plugins refer to the plugins were 7 loaded when needed due its bulky and/or memory hungry dependencies. 8 Those plugins has to use lml and respect lml's design principle. 9 10 load-me-now plugins refer to the plugins are immediately imported. All 11 conventional Python classes are by default immediately imported. 12 13 :class:`~lml.plugin.PluginManager` should be inherited to form new 14 plugin manager class. If you have more than one plugins in your 15 architecture, it is advisable to have one class per plugin type. 16 17 :class:`~lml.plugin.PluginInfoChain` helps the plugin module to 18 declare the available plugins in the module. 19 20 :class:`~lml.plugin.PluginInfo` can be subclassed to describe 21 your plugin. Its method :meth:`~lml.plugin.PluginInfo.tags` 22 can be overridden to help its matching :class:`~lml.plugin.PluginManager` 23 to look itself up. 24 25 :copyright: (c) 2017-2020 by Onni Software Ltd. 26 :license: New BSD License, see LICENSE for more details 27""" 28import logging 29from collections import defaultdict 30 31from lml.utils import json_dumps, do_import_class 32 33PLUG_IN_MANAGERS = {} 34CACHED_PLUGIN_INFO = defaultdict(list) 35 36log = logging.getLogger(__name__) 37 38 39class PluginInfo(object): 40 """ 41 Information about the plugin. 42 43 It is used together with PluginInfoChain to describe the plugins. 44 Meanwhile, it is a class decorator and can be used to register a plugin 45 immediately for use, in other words, the PluginInfo decorated plugin 46 class is not loaded later. 47 48 Parameters 49 ------------- 50 name: 51 plugin name 52 53 absolute_import_path: 54 absolute import path from your plugin name space for your plugin class 55 56 tags: 57 a list of keywords help the plugin manager to retrieve your plugin 58 59 keywords: 60 Another custom properties. 61 62 Examples 63 ------------- 64 65 For load-me-later plugins: 66 67 >>> info = PluginInfo("sample", 68 ... abs_class_path='lml.plugin.PluginInfo', # demonstration only. 69 ... tags=['load-me-later'], 70 ... custom_property = 'I am a custom property') 71 >>> print(info.module_name) 72 lml 73 >>> print(info.custom_property) 74 I am a custom property 75 76 For load-me-now plugins: 77 78 >>> @PluginInfo("sample", tags=['load-me-now']) 79 ... class TestPlugin: 80 ... def echo(self, words): 81 ... print("echoing %s" % words) 82 83 Now let's retrive the second plugin back: 84 85 >>> class SamplePluginManager(PluginManager): 86 ... def __init__(self): 87 ... PluginManager.__init__(self, "sample") 88 >>> sample_manager = SamplePluginManager() 89 >>> test_plugin=sample_manager.get_a_plugin("load-me-now") 90 >>> test_plugin.echo("hey..") 91 echoing hey.. 92 93 """ 94 95 def __init__( 96 self, plugin_type, abs_class_path=None, tags=None, **keywords 97 ): 98 self.plugin_type = plugin_type 99 self.absolute_import_path = abs_class_path 100 self.cls = None 101 self.properties = keywords 102 self.__tags = tags 103 104 def __getattr__(self, name): 105 if name == "module_name": 106 if self.absolute_import_path: 107 module_name = self.absolute_import_path.split(".")[0] 108 else: 109 module_name = self.cls.__module__ 110 return module_name 111 return self.properties.get(name) 112 113 def tags(self): 114 """ 115 A list of tags for identifying the plugin class 116 117 The plugin class is described at the absolute_import_path 118 """ 119 if self.__tags is None: 120 yield self.plugin_type 121 else: 122 for tag in self.__tags: 123 yield tag 124 125 def __repr__(self): 126 rep = { 127 "plugin_type": self.plugin_type, 128 "path": self.absolute_import_path, 129 } 130 rep.update(self.properties) 131 return json_dumps(rep) 132 133 def __call__(self, cls): 134 self.cls = cls 135 _register_a_plugin(self, cls) 136 return cls 137 138 139class PluginInfoChain(object): 140 """ 141 Pandas style, chained list declaration 142 143 It is used in the plugin packages to list all plugin classes 144 """ 145 146 def __init__(self, path): 147 self._logger = logging.getLogger( 148 self.__class__.__module__ + "." + self.__class__.__name__ 149 ) 150 self.module_name = path 151 152 def add_a_plugin(self, plugin_type, submodule=None, **keywords): 153 """ 154 Add a plain plugin 155 156 Parameters 157 ------------- 158 159 plugin_type: 160 plugin manager name 161 162 submodule: 163 the relative import path to your plugin class 164 """ 165 a_plugin_info = PluginInfo( 166 plugin_type, self._get_abs_path(submodule), **keywords 167 ) 168 169 self.add_a_plugin_instance(a_plugin_info) 170 return self 171 172 def add_a_plugin_instance(self, plugin_info_instance): 173 """ 174 Add a plain plugin 175 176 Parameters 177 ------------- 178 179 plugin_info_instance: 180 an instance of PluginInfo 181 182 The developer has to specify the absolute import path 183 """ 184 self._logger.debug( 185 "add %s as '%s' plugin", 186 plugin_info_instance.absolute_import_path, 187 plugin_info_instance.plugin_type, 188 ) 189 _load_me_later(plugin_info_instance) 190 return self 191 192 def _get_abs_path(self, submodule): 193 return "%s.%s" % (self.module_name, submodule) 194 195 196class PluginManager(object): 197 """ 198 Load plugin info into in-memory dictionary for later import 199 200 Parameters 201 -------------- 202 203 plugin_type: 204 the plugin type. All plugins of this plugin type will be 205 registered to it. 206 """ 207 208 def __init__(self, plugin_type): 209 self.plugin_name = plugin_type 210 self.registry = defaultdict(list) 211 self.tag_groups = dict() 212 self._logger = logging.getLogger( 213 self.__class__.__module__ + "." + self.__class__.__name__ 214 ) 215 _register_class(self) 216 217 def get_a_plugin(self, key, **keywords): 218 """ Get a plugin 219 220 Parameters 221 --------------- 222 223 key: 224 the key to find the plugins 225 226 keywords: 227 additional parameters for help the retrieval of the plugins 228 """ 229 self._logger.debug("get a plugin called") 230 plugin = self.load_me_now(key) 231 return plugin() 232 233 def raise_exception(self, key): 234 """Raise plugin not found exception 235 236 Override this method to raise custom exception 237 238 Parameters 239 ----------------- 240 241 key: 242 the key to find the plugin 243 """ 244 self._logger.debug(self.registry.keys()) 245 raise Exception("No %s is found for %s" % (self.plugin_name, key)) 246 247 def load_me_later(self, plugin_info): 248 """ 249 Register a plugin info for later loading 250 251 Parameters 252 -------------- 253 254 plugin_info: 255 a instance of plugin info 256 """ 257 self._logger.debug("load %s later", plugin_info.absolute_import_path) 258 self._update_registry_and_expand_tag_groups(plugin_info) 259 260 def load_me_now(self, key, library=None, **keywords): 261 """ 262 Import a plugin from plugin registry 263 264 Parameters 265 ----------------- 266 267 key: 268 the key to find the plugin 269 270 library: 271 to use a specific plugin module 272 """ 273 if keywords: 274 self._logger.debug(keywords) 275 __key = key.lower() 276 277 if __key in self.registry: 278 for plugin_info in self.registry[__key]: 279 cls = self.dynamic_load_library(plugin_info) 280 module_name = _get_me_pypi_package_name(cls) 281 if library and module_name != library: 282 continue 283 else: 284 break 285 else: 286 # only library condition could raise an exception 287 self._logger.debug("%s is not installed" % library) 288 self.raise_exception(key) 289 self._logger.debug("load %s now for '%s'", cls, key) 290 return cls 291 else: 292 self.raise_exception(key) 293 294 def dynamic_load_library(self, a_plugin_info): 295 """Dynamically load the plugin info if not loaded 296 297 298 Parameters 299 -------------- 300 301 a_plugin_info: 302 a instance of plugin info 303 """ 304 if a_plugin_info.cls is None: 305 self._logger.debug("import " + a_plugin_info.absolute_import_path) 306 cls = do_import_class(a_plugin_info.absolute_import_path) 307 a_plugin_info.cls = cls 308 return a_plugin_info.cls 309 310 def register_a_plugin(self, plugin_cls, plugin_info): 311 """ for dynamically loaded plugin during runtime 312 313 Parameters 314 -------------- 315 316 plugin_cls: 317 the actual plugin class refered to by the second parameter 318 319 plugin_info: 320 a instance of plugin info 321 """ 322 self._logger.debug("register %s", _show_me_your_name(plugin_cls)) 323 plugin_info.cls = plugin_cls 324 self._update_registry_and_expand_tag_groups(plugin_info) 325 326 def get_primary_key(self, key): 327 __key = key.lower() 328 return self.tag_groups.get(__key, None) 329 330 def _update_registry_and_expand_tag_groups(self, plugin_info): 331 primary_tag = None 332 for index, key in enumerate(plugin_info.tags()): 333 self.registry[key.lower()].append(plugin_info) 334 if index == 0: 335 primary_tag = key.lower() 336 self.tag_groups[key.lower()] = primary_tag 337 338 339def _register_class(cls): 340 """Reigister a newly created plugin manager""" 341 log.debug("declare '%s' plugin manager", cls.plugin_name) 342 PLUG_IN_MANAGERS[cls.plugin_name] = cls 343 if cls.plugin_name in CACHED_PLUGIN_INFO: 344 # check if there is early registrations or not 345 for plugin_info in CACHED_PLUGIN_INFO[cls.plugin_name]: 346 if plugin_info.absolute_import_path: 347 log.debug( 348 "load cached plugin info: %s", 349 plugin_info.absolute_import_path, 350 ) 351 else: 352 log.debug( 353 "load cached plugin info: %s", 354 _show_me_your_name(plugin_info.cls), 355 ) 356 cls.load_me_later(plugin_info) 357 358 del CACHED_PLUGIN_INFO[cls.plugin_name] 359 360 361def _register_a_plugin(plugin_info, plugin_cls): 362 """module level function to register a plugin""" 363 manager = PLUG_IN_MANAGERS.get(plugin_info.plugin_type) 364 if manager: 365 manager.register_a_plugin(plugin_cls, plugin_info) 366 else: 367 # let's cache it and wait the manager to be registered 368 try: 369 log.debug("caching %s", _show_me_your_name(plugin_cls.__name__)) 370 except AttributeError: 371 log.debug("caching %s", _show_me_your_name(plugin_cls)) 372 CACHED_PLUGIN_INFO[plugin_info.plugin_type].append(plugin_info) 373 374 375def _load_me_later(plugin_info): 376 """ module level function to load a plugin later""" 377 manager = PLUG_IN_MANAGERS.get(plugin_info.plugin_type) 378 if manager: 379 manager.load_me_later(plugin_info) 380 else: 381 # let's cache it and wait the manager to be registered 382 log.debug( 383 "caching %s for %s", 384 plugin_info.absolute_import_path, 385 plugin_info.plugin_type, 386 ) 387 CACHED_PLUGIN_INFO[plugin_info.plugin_type].append(plugin_info) 388 389 390def _get_me_pypi_package_name(module): 391 try: 392 module_name = module.__module__ 393 root_module_name = module_name.split(".")[0] 394 return root_module_name.replace("_", "-") 395 except AttributeError: 396 return None 397 398 399def _show_me_your_name(cls_func_or_data_type): 400 try: 401 return cls_func_or_data_type.__name__ 402 except AttributeError: 403 return str(type(cls_func_or_data_type)) 404