Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/lml/plugin.py : 47%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2 lml.plugin
3 ~~~~~~~~~~~~~~~~~~~
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.
10 load-me-now plugins refer to the plugins are immediately imported. All
11 conventional Python classes are by default immediately imported.
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.
17 :class:`~lml.plugin.PluginInfoChain` helps the plugin module to
18 declare the available plugins in the module.
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.
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
31from lml.utils import json_dumps, do_import_class
33PLUG_IN_MANAGERS = {}
34CACHED_PLUGIN_INFO = defaultdict(list)
36log = logging.getLogger(__name__)
39class PluginInfo(object):
40 """
41 Information about the plugin.
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.
48 Parameters
49 -------------
50 name:
51 plugin name
53 absolute_import_path:
54 absolute import path from your plugin name space for your plugin class
56 tags:
57 a list of keywords help the plugin manager to retrieve your plugin
59 keywords:
60 Another custom properties.
62 Examples
63 -------------
65 For load-me-later plugins:
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
76 For load-me-now plugins:
78 >>> @PluginInfo("sample", tags=['load-me-now'])
79 ... class TestPlugin:
80 ... def echo(self, words):
81 ... print("echoing %s" % words)
83 Now let's retrive the second plugin back:
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..
93 """
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
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)
113 def tags(self):
114 """
115 A list of tags for identifying the plugin class
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
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)
133 def __call__(self, cls):
134 self.cls = cls
135 _register_a_plugin(self, cls)
136 return cls
139class PluginInfoChain(object):
140 """
141 Pandas style, chained list declaration
143 It is used in the plugin packages to list all plugin classes
144 """
146 def __init__(self, path):
147 self._logger = logging.getLogger(
148 self.__class__.__module__ + "." + self.__class__.__name__
149 )
150 self.module_name = path
152 def add_a_plugin(self, plugin_type, submodule=None, **keywords):
153 """
154 Add a plain plugin
156 Parameters
157 -------------
159 plugin_type:
160 plugin manager name
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 )
169 self.add_a_plugin_instance(a_plugin_info)
170 return self
172 def add_a_plugin_instance(self, plugin_info_instance):
173 """
174 Add a plain plugin
176 Parameters
177 -------------
179 plugin_info_instance:
180 an instance of PluginInfo
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
192 def _get_abs_path(self, submodule):
193 return "%s.%s" % (self.module_name, submodule)
196class PluginManager(object):
197 """
198 Load plugin info into in-memory dictionary for later import
200 Parameters
201 --------------
203 plugin_type:
204 the plugin type. All plugins of this plugin type will be
205 registered to it.
206 """
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)
217 def get_a_plugin(self, key, **keywords):
218 """ Get a plugin
220 Parameters
221 ---------------
223 key:
224 the key to find the plugins
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()
233 def raise_exception(self, key):
234 """Raise plugin not found exception
236 Override this method to raise custom exception
238 Parameters
239 -----------------
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))
247 def load_me_later(self, plugin_info):
248 """
249 Register a plugin info for later loading
251 Parameters
252 --------------
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)
260 def load_me_now(self, key, library=None, **keywords):
261 """
262 Import a plugin from plugin registry
264 Parameters
265 -----------------
267 key:
268 the key to find the plugin
270 library:
271 to use a specific plugin module
272 """
273 if keywords:
274 self._logger.debug(keywords)
275 __key = key.lower()
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)
294 def dynamic_load_library(self, a_plugin_info):
295 """Dynamically load the plugin info if not loaded
298 Parameters
299 --------------
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
310 def register_a_plugin(self, plugin_cls, plugin_info):
311 """ for dynamically loaded plugin during runtime
313 Parameters
314 --------------
316 plugin_cls:
317 the actual plugin class refered to by the second parameter
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)
326 def get_primary_key(self, key):
327 __key = key.lower()
328 return self.tag_groups.get(__key, None)
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
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)
358 del CACHED_PLUGIN_INFO[cls.plugin_name]
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)
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)
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
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))