Source code for lml.plugin


    lml divides the plugins into two category: load-me-later plugins and
    load-me-now ones. load-me-later plugins refer to the plugins were
    loaded when needed due its bulky and/or memory hungry dependencies.
    Those plugins has to use lml and respect lml's design principle.

    load-me-now plugins refer to the plugins are immediately imported. All
    conventional Python classes are by default immediately imported.

    :class:`~lml.plugin.PluginManager` should be inherited to form new
    plugin manager class. If you have more than one plugins in your
    architecture, it is advisable to have one class per plugin type.

    :class:`~lml.plugin.PluginInfoChain` helps the plugin module to
    declare the available plugins in the module.

    :class:`~lml.plugin.PluginInfo` can be subclassed to describe
    your plugin. Its method :meth:`~lml.plugin.PluginInfo.tags`
    can be overridden to help its matching :class:`~lml.plugin.PluginManager`
    to look itself up.

    :copyright: (c) 2017-2020 by Onni Software Ltd.
    :license: New BSD License, see LICENSE for more details
import logging
from collections import defaultdict

from lml.utils import json_dumps, do_import_class

CACHED_PLUGIN_INFO = defaultdict(list)

log = logging.getLogger(__name__)

[docs]class PluginInfo(object): """ Information about the plugin. It is used together with PluginInfoChain to describe the plugins. Meanwhile, it is a class decorator and can be used to register a plugin immediately for use, in other words, the PluginInfo decorated plugin class is not loaded later. Parameters ------------- name: plugin name absolute_import_path: absolute import path from your plugin name space for your plugin class tags: a list of keywords help the plugin manager to retrieve your plugin keywords: Another custom properties. Examples ------------- For load-me-later plugins: >>> info = PluginInfo("sample", ... abs_class_path='lml.plugin.PluginInfo', # demonstration only. ... tags=['load-me-later'], ... custom_property = 'I am a custom property') >>> print(info.module_name) lml >>> print(info.custom_property) I am a custom property For load-me-now plugins: >>> @PluginInfo("sample", tags=['load-me-now']) ... class TestPlugin: ... def echo(self, words): ... print("echoing %s" % words) Now let's retrive the second plugin back: >>> class SamplePluginManager(PluginManager): ... def __init__(self): ... PluginManager.__init__(self, "sample") >>> sample_manager = SamplePluginManager() >>> test_plugin=sample_manager.get_a_plugin("load-me-now") >>> test_plugin.echo("hey..") echoing hey.. """ def __init__( self, plugin_type, abs_class_path=None, tags=None, **keywords ): self.plugin_type = plugin_type self.absolute_import_path = abs_class_path self.cls = None = keywords self.__tags = tags def __getattr__(self, name): if name == "module_name": if self.absolute_import_path: module_name = self.absolute_import_path.split(".")[0] else: module_name = self.cls.__module__ return module_name return def tags(self): """ A list of tags for identifying the plugin class The plugin class is described at the absolute_import_path """ if self.__tags is None: yield self.plugin_type else: for tag in self.__tags: yield tag def __repr__(self): rep = { "plugin_type": self.plugin_type, "path": self.absolute_import_path, } rep.update( return json_dumps(rep) def __call__(self, cls): self.cls = cls _register_a_plugin(self, cls) return cls
[docs]class PluginInfoChain(object): """ Pandas style, chained list declaration It is used in the plugin packages to list all plugin classes """ def __init__(self, path): self._logger = logging.getLogger( self.__class__.__module__ + "." + self.__class__.__name__ ) self.module_name = path def add_a_plugin(self, plugin_type, submodule=None, **keywords): """ Add a plain plugin Parameters ------------- plugin_type: plugin manager name submodule: the relative import path to your plugin class """ a_plugin_info = PluginInfo( plugin_type, self._get_abs_path(submodule), **keywords ) self.add_a_plugin_instance(a_plugin_info) return self def add_a_plugin_instance(self, plugin_info_instance): """ Add a plain plugin Parameters ------------- plugin_info_instance: an instance of PluginInfo The developer has to specify the absolute import path """ self._logger.debug( "add %s as '%s' plugin", plugin_info_instance.absolute_import_path, plugin_info_instance.plugin_type, ) _load_me_later(plugin_info_instance) return self def _get_abs_path(self, submodule): return "%s.%s" % (self.module_name, submodule)
[docs]class PluginManager(object): """ Load plugin info into in-memory dictionary for later import Parameters -------------- plugin_type: the plugin type. All plugins of this plugin type will be registered to it. """ def __init__(self, plugin_type): self.plugin_name = plugin_type self.registry = defaultdict(list) self.tag_groups = dict() self._logger = logging.getLogger( self.__class__.__module__ + "." + self.__class__.__name__ ) _register_class(self)
[docs] def get_a_plugin(self, key, **keywords): """ Get a plugin Parameters --------------- key: the key to find the plugins keywords: additional parameters for help the retrieval of the plugins """ self._logger.debug("get a plugin called") plugin = self.load_me_now(key) return plugin()
[docs] def raise_exception(self, key): """Raise plugin not found exception Override this method to raise custom exception Parameters ----------------- key: the key to find the plugin """ self._logger.debug(self.registry.keys()) raise Exception("No %s is found for %s" % (self.plugin_name, key))
[docs] def load_me_later(self, plugin_info): """ Register a plugin info for later loading Parameters -------------- plugin_info: a instance of plugin info """ self._logger.debug("load %s later", plugin_info.absolute_import_path) self._update_registry_and_expand_tag_groups(plugin_info)
[docs] def load_me_now(self, key, library=None, **keywords): """ Import a plugin from plugin registry Parameters ----------------- key: the key to find the plugin library: to use a specific plugin module """ if keywords: self._logger.debug(keywords) __key = key.lower() if __key in self.registry: for plugin_info in self.registry[__key]: cls = self.dynamic_load_library(plugin_info) module_name = _get_me_pypi_package_name(cls) if library and module_name != library: continue else: break else: # only library condition could raise an exception self._logger.debug("%s is not installed" % library) self.raise_exception(key) self._logger.debug("load %s now for '%s'", cls, key) return cls else: self.raise_exception(key)
[docs] def dynamic_load_library(self, a_plugin_info): """Dynamically load the plugin info if not loaded Parameters -------------- a_plugin_info: a instance of plugin info """ if a_plugin_info.cls is None: self._logger.debug("import " + a_plugin_info.absolute_import_path) cls = do_import_class(a_plugin_info.absolute_import_path) a_plugin_info.cls = cls return a_plugin_info.cls
[docs] def register_a_plugin(self, plugin_cls, plugin_info): """ for dynamically loaded plugin during runtime Parameters -------------- plugin_cls: the actual plugin class refered to by the second parameter plugin_info: a instance of plugin info """ self._logger.debug("register %s", _show_me_your_name(plugin_cls)) plugin_info.cls = plugin_cls self._update_registry_and_expand_tag_groups(plugin_info)
def get_primary_key(self, key): __key = key.lower() return self.tag_groups.get(__key, None) def _update_registry_and_expand_tag_groups(self, plugin_info): primary_tag = None for index, key in enumerate(plugin_info.tags()): self.registry[key.lower()].append(plugin_info) if index == 0: primary_tag = key.lower() self.tag_groups[key.lower()] = primary_tag
def _register_class(cls): """Reigister a newly created plugin manager""" log.debug("declare '%s' plugin manager", cls.plugin_name) PLUG_IN_MANAGERS[cls.plugin_name] = cls if cls.plugin_name in CACHED_PLUGIN_INFO: # check if there is early registrations or not for plugin_info in CACHED_PLUGIN_INFO[cls.plugin_name]: if plugin_info.absolute_import_path: log.debug( "load cached plugin info: %s", plugin_info.absolute_import_path, ) else: log.debug( "load cached plugin info: %s", _show_me_your_name(plugin_info.cls), ) cls.load_me_later(plugin_info) del CACHED_PLUGIN_INFO[cls.plugin_name] def _register_a_plugin(plugin_info, plugin_cls): """module level function to register a plugin""" manager = PLUG_IN_MANAGERS.get(plugin_info.plugin_type) if manager: manager.register_a_plugin(plugin_cls, plugin_info) else: # let's cache it and wait the manager to be registered try: log.debug("caching %s", _show_me_your_name(plugin_cls.__name__)) except AttributeError: log.debug("caching %s", _show_me_your_name(plugin_cls)) CACHED_PLUGIN_INFO[plugin_info.plugin_type].append(plugin_info) def _load_me_later(plugin_info): """ module level function to load a plugin later""" manager = PLUG_IN_MANAGERS.get(plugin_info.plugin_type) if manager: manager.load_me_later(plugin_info) else: # let's cache it and wait the manager to be registered log.debug( "caching %s for %s", plugin_info.absolute_import_path, plugin_info.plugin_type, ) CACHED_PLUGIN_INFO[plugin_info.plugin_type].append(plugin_info) def _get_me_pypi_package_name(module): try: module_name = module.__module__ root_module_name = module_name.split(".")[0] return root_module_name.replace("_", "-") except AttributeError: return None def _show_me_your_name(cls_func_or_data_type): try: return cls_func_or_data_type.__name__ except AttributeError: return str(type(cls_func_or_data_type))