Source code for openpaperwork_core

import collections
import importlib
import logging
import os
import time


LOGGER = logging.getLogger(__name__)


[docs]class DependencyException(Exception): """ Failed to satisfy dependencies. """ pass
[docs]class PluginBase(object): """ Indicates all the methods that must be implemented by any plugin managed by OpenPaperwork core. Also provides default implementations for each method. """ # Priority defines in which order callbacks will be called. # Plugins with higher priorities will have their callbacks called first. PRIORITY = 0 # Convenience for the applications: Indicates if users should be able # to enable/disable this plugin in the UI. USER_VISIBLE = False def __init__(self): """ Called as soon as the module is loaded. Should be as minimal as possible. Most of the work should be done in `init()`. You *must* *not* rely on any dependencies here. """ self.core = None
[docs] def get_interfaces(self): """ Indicates the list of interfaces implemented by this plugin. Interface names are arbitrarily defined. Methods provided by each interface are arbitrarily defined (and no checks are done). Returns a list of string. """ return []
[docs] def get_deps(self): """ Return the dependencies required by this plugin. Example: .. code-block:: python [ { "interface": "some_interface_name", # required "defaults": ['plugin_a', 'plugin_b'], # required "expected_already_satisfied": False, # optional, default: True }, ] """ return []
[docs] def init(self, core): """ Plugins can initialize whatever they want here. When called, all dependencies have been loaded and initialized, so using them is safe. Does nothing by default. """ self.core = core
[docs]class Core(object): """ Manage plugins and their callbacks. """ def __init__(self, allow_unsatisfied=False): """ `allow_unsatisfied=True` means that missing dependencies will be loaded automatically based on the default plugin list provided by plugins. This should be only used for testing. """ self.plugins = {} self.initialized = False self._to_initialize = set() self._initialized = set() # avoid double-init self.interfaces = collections.defaultdict(list) self.callbacks = collections.defaultdict(list) self.allow_unsatisfied = allow_unsatisfied self.log_all = bool(os.getenv("CORE_LOG_ALL", 0)) self.count_limit_per_second = int(os.getenv("CORE_CALL_LIMIT", 0)) self.counters_last_reset = 0 self.counters = collections.defaultdict(lambda: 0)
[docs] def load(self, module_name): """ - Load the specified module - Instantiate the class 'Plugin()' of this module - Register all the methods of this plugin object (except those starting by '_' and those from the class PluginBase) as callbacks BEWARE of dependency loops ! Arguments: - module_name: name of the Python module to load """ if module_name in self.plugins: return self.plugins[module_name] LOGGER.info("Loading plugin '%s' ...", module_name) try: module = importlib.import_module(module_name) return self._load_module(module_name, module) except Exception as exc: LOGGER.error("Failed to load '%s'", module_name, exc_info=exc) return None
def _load_module(self, module_name, module): """ should be called from outside for testing only """ if module_name in self.plugins: LOGGER.debug("Module %s already loaded", module_name) return self.plugins[module_name] self.initialized = False plugin = module.Plugin() self.plugins[module_name] = plugin for interface in plugin.get_interfaces(): LOGGER.debug("- '%s' provides '%s'", str(type(plugin)), interface) self.interfaces[interface].append(plugin) self._to_initialize.add(plugin) LOGGER.info("Plugin '%s' loaded", module_name) return plugin def _check_deps(self): to_examine = [ (plugin_name, plugin) for (plugin_name, plugin) in self.plugins.items() ] while len(to_examine) > 0: (plugin_name, plugin) = to_examine[0] to_examine = to_examine[1:] LOGGER.info("Examining dependencies of '%s' ...", plugin_name) deps = plugin.get_deps() for dep in deps: interface = dep['interface'] if len(self.interfaces[interface]) > 0: LOGGER.debug( "- Interface '%s' already provided by %d plugins", interface, len(self.interfaces[interface]) ) continue defaults = dep['defaults'] if len(defaults) <= 0: continue if (not self.allow_unsatisfied and ( 'expected_already_satisfied' not in dep or dep['expected_already_satisfied'] )): LOGGER.warning( "Plugin '{}' requires interface '{}' but no plugins" " provide this interface (suggested: {}). Plugin '{}'" " will not be initialized.".format( plugin_name, interface, defaults, plugin_name ) ) plugin = self.plugins.pop(plugin_name) self._to_initialize.remove(plugin) # return False to indicate we actually dropped a plugin # and need to reevaluate all the dependencies again. return False else: LOGGER.info( "Loading plugins %s to satisfy dependency." " Required by '%s' for interface '%s'", defaults, type(plugin), interface ) for default in defaults: to_examine.append((default, self.load(default))) return True def _register_plugin(self, plugin): for attr_name in dir(plugin): if attr_name[0] == "_": continue if attr_name in dir(PluginBase): # ignore base methods of plugins continue callback = getattr(plugin, attr_name) if not hasattr(callback, '__call__'): continue LOGGER.debug("- %s.%s()", str(type(plugin)), attr_name) self.callbacks[attr_name].append(( plugin.PRIORITY, str(type(plugin)), callback )) self.callbacks[attr_name].sort(reverse=True) def _init(self, plugin): nb = 0 if plugin in self._initialized: return nb self.initialized = True deps = plugin.get_deps() for dep in deps: dep_plugins = self.interfaces[dep['interface']] for dep_plugin in dep_plugins: nb += self._init(dep_plugin) LOGGER.info("Initializing plugin '%s' ...", type(plugin)) try: plugin.init(self) except Exception as exc: LOGGER.error( "Failed to initialized plugin '%s'", type(plugin), exc_info=exc ) return nb self._register_plugin(plugin) nb += 1 self._initialized.add(plugin) return nb
[docs] def init(self): """ - Make sure all the dependencies of all the plugins are satisfied. - Call the method init() of each plugin following the dependency order (those without dependencies are called first). BEWARE of dependency loops ! """ LOGGER.info("Initializing all plugins") while not self._check_deps(): pass nb = 0 for plugin in self._to_initialize: nb += self._init(plugin) self._to_initialize = set() LOGGER.info("%d plugins initialized", nb)
[docs] def get_by_name(self, module_name): """ Returns a Plugin instance based on the corresponding module name (assuming it has been loaded). This method is here for unit testing only. You shouldn't use it in production code. """ return self.plugins[module_name]
[docs] def get_by_interface(self, interface_name): return self.interfaces[interface_name]
def _check_call_limit(self, callback_name): if self.count_limit_per_second <= 0: return now = time.time() if now - self.counters_last_reset >= 1.0: self.counters = collections.defaultdict(lambda: 0) self.counters_last_reset = now self.counters[callback_name] += 1 if self.counters[callback_name] >= self.count_limit_per_second: raise Exception( "Too many calls to '{}' (>= {}) in one second".format( callback_name, self.count_limit_per_second ) )
[docs] def call_all(self, callback_name, *args, **kwargs): """ Call all the methods of all the plugins that have `callback_name` as name. Arguments are passed as is. Returned values are dropped (use callbacks for return values if required) Method call order is defined by the plugin priorities: Plugins with a higher priority get their methods called first. When we need a return value from callbacks called with `call_all()`, we need a way to get the results from all of them. The usual way to do that is to instantiate an empty `list` or `set`, and pass it as first argument of the callbacks (argument `out`). Callbacks can then complete this list or set using `list.append()` or `set.add()`. .. uml:: Caller -> Core: call "func" Core -> "Plugin A": plugin.func() Core <- "Plugin A": returns "something_a" Core -> "Plugin B": plugin.func() Core <- "Plugin B": returns "something_b" Core -> "Plugin C": plugin.func() Core <- "Plugin C": returns "something_c" Caller <- Core: returns 3 """ if self.log_all: print( "[{}] call_all({}, args={}, kwargs={})".format( time.time(), callback_name, args, kwargs ) ) assert \ self.initialized, \ "A plugin has been loaded without being initialized." \ " Call core.init() first" self._check_call_limit(callback_name) callbacks = self.callbacks[callback_name] if len(callbacks) <= 0: if callback_name.startswith("on_"): # those are 'observer' callback. If nobody is observing, # it's usually fine. log_method = LOGGER.debug else: log_method = LOGGER.warning log_method("No method '%s' found", callback_name) return 0 for (priority, plugin, callback) in callbacks: if self.log_all: print( "[{}] call_all({}, args={}, kwargs={}) -> {}:{}".format( time.time(), callback_name, args, kwargs, priority, callback ) ) callback(*args, **kwargs) return len(callbacks)
[docs] def call_one(self, callback_name, *args, **kwargs): """ Look for a plugin method called `callback_name` and calls it. Raises an error if no such method exists. If many exists, raises a warning and call one at random. Returns the value return by the callback. Method call order is defined by the plugin priorities: Plugins with a higher priority get their methods called first. .. uml:: Caller -> Core: call "func" Core -> "Plugin A": plugin.func() Core <- "Plugin A": returns X Caller <- Core: returns X You're advised to use `call_all()` or `call_success()` instead whenever possible. This method is only provided as convenience for when you're fairly sure there should be only one plugin with such callback (example: mainloop plugins). """ assert \ self.initialized, \ "A plugin has been loaded without being initialized." \ " Call core.init() first" self._check_call_limit(callback_name) if self.log_all: print( "[{}] call_one({}, args={}, kwargs={})".format( time.time(), callback_name, args, kwargs ) ) callbacks = self.callbacks[callback_name] if len(callbacks) <= 0: raise IndexError( "No method '{}' found !".format(callback_name) ) if len(callbacks) > 1: LOGGER.warning( "More than one method '%s' found ! [%s]", callback_name, ", ".join([callback[1] for callback in callbacks]) ) if self.log_all: print( "[{}] call_all({}, args={}, kwargs={}) -> {}:{}".format( time.time(), callback_name, args, kwargs, callbacks[0][0], callbacks[0][2] ) ) return callbacks[0][2](*args, **kwargs)
[docs] def call_success(self, callback_name, *args, **kwargs): """ Call methods of all the plugins that have `callback_name` as name until one of them return a value that is not None. Arguments are passed as is. First value to be different from None is returned. If none of the callbacks returned a value different from None or if no callback has the specified name, this method will return None. Method call order is defined by the plugin priorities: Plugins with a higher priority get their methods called first. Callbacks should never raise any exception. .. uml:: Caller -> Core: call "func" Core -> "Plugin A": plugin.func() Core <- "Plugin A": returns None Core -> "Plugin B": plugin.func() Core <- "Plugin B": returns None Core -> "Plugin C": plugin.func() Core <- "Plugin C": returns "something" Caller <- Core: returns "something" """ assert \ self.initialized, \ "A plugin has been loaded without being initialized." \ " Call core.init() first" self._check_call_limit(callback_name) if self.log_all: print( "[{}] call_one({}, args={}, kwargs={})".format( time.time(), callback_name, args, kwargs ) ) callbacks = self.callbacks[callback_name] if len(callbacks) <= 0: LOGGER.warning("No method '%s' found", callback_name) for (priority, plugin, callback) in callbacks: if self.log_all: print( "[{}] call_all({}, args={}, kwargs={}) -> {}:{}".format( time.time(), callback_name, args, kwargs, priority, callback ) ) try: r = callback(*args, **kwargs) except Exception: LOGGER.error("Callback '%s' failed", str(callback)) raise if r is not None: return r return None