Source code for compas.plugins


# The COMPAS plugin system owes a lot to pluggy, the pytest plugin framework
# There are portions of code loosely based on pluggy's
# and while it is not strictly derivative work, we include
# their license and copyright notice to give credit where credit is due.
#
# MIT license
# Copyright (c) 2015 holger krekel (rather uses bitbucket/hpk42)
#
# https://github.com/pytest-dev/pluggy
"""
********************************************************************************
plugins
********************************************************************************

.. currentmodule:: compas.plugins

COMPAS has an extensible architecture based on plugins that allows to
customize and extend the functionality of the core framework.

Check out the developer guide to :ref:`plugins` for additional details.

Decorators
==========

.. autosummary::
    :toctree: generated/
    :nosignatures:

    pluggable
    plugin

Classes
=======

.. autosummary::
    :toctree: generated/
    :nosignatures:

    PluginManager

Exceptions
==========

.. autosummary::
    :toctree: generated/
    :nosignatures:

    PluginNotInstalledError
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import functools
import inspect
import pkgutil
import threading

__all__ = [
    'pluggable',
    'plugin',
    'plugin_manager',
    'PluginManager',
    'PluginNotInstalledError',
]


[docs]class PluginNotInstalledError(Exception): """Exception raised when an extension point is invoked but no plugin is available.""" pass
def _get_extension_point_url_from_name(domain, category, pluggable_name): """Get the extension point URL based on a pluggable method name""" return '{}/{}/{}'.format(domain, category, pluggable_name).replace('//', '/') def _get_extension_point_url_from_method(domain, category, plugin_method): """Get the extension point URL based on a method instance""" name = getattr(plugin_method, '__name__', None) or str(id(plugin_method)) return '{}/{}/{}'.format(domain, category, name).replace('//', '/') class PluginImpl(object): """Internal data class to keep track of a loaded plugin implementation. Parameters ---------- plugin : module Instance of the module containing one or more plugin implementations. method : method Method implementing the a plugin's behavior. plugin_opts : dict Dictionary containing plugin options. """ def __init__(self, plugin, method, plugin_opts): self.plugin = plugin self.method = method self.opts = plugin_opts if plugin_opts['tryfirst']: self.key = 1 elif plugin_opts['trylast']: self.key = 3 else: self.key = 2 @property def id(self): """Identifier of the plugin implementation.""" return '{}.{}'.format(self.plugin.__name__, self.method.__name__) def __repr__(self): return '<PluginImpl id={}, plugin_module={}>'.format(self.id, self.plugin)
[docs]class PluginManager(object): """Plugin Manager handles discovery and registry of plugins. Usually there is only one instance of a plugin manager per host. """ DEBUG = False
[docs] def __init__(self): self.importer = Importer() self._registry = {} self._discovery_done = False self._discovery_lock = threading.Lock()
@property def registry(self): """Plugin registry. Lazy-loaded dictionary of all plugins available in the system. Returns ------- dict Dictionary of available plugins. The keys are extension point URLs and the values are instances of :class:`PluginImpl`. """ if not self._discovery_done: self.load_plugins() return self._registry
[docs] def load_plugins(self): """Load available plugin modules. Returns ------- int Number of loaded plugins. """ # Since we modify global state, # let's lock around this. with self._discovery_lock: count = 0 modules = [ module_name for _importer, module_name, is_pkg in pkgutil.iter_modules() if is_pkg and module_name.startswith('compas') ] modules_to_inspect = dict() for module_name in modules: module = self.importer.try_import(module_name) if module: modules_to_inspect[module_name] = module else: if self.DEBUG: print('Error importing module {}, skipping entire package.'.format(module_name)) continue if '__all_plugins__' in dir(module): for plugin_module_name in module.__all_plugins__: plugin_module = self.importer.try_import(plugin_module_name) if plugin_module: modules_to_inspect[plugin_module_name] = plugin_module else: if self.DEBUG: print('Error importing plugin {}, skipping.'.format(plugin_module_name)) if self.DEBUG: print('Will inspect modules: {}'.format(list(modules_to_inspect.keys()))) for plugin_module in modules_to_inspect.values(): count += self.register_module(plugin_module) self._discovery_done = True return count
[docs] def register_module(self, plugin_module): """Register a module that potentially contains plugin implementations. Parameters ---------- plugin_module : module Module instance to inspect for plugins. Returns ------- int Count of successfully registered plugins in the module. """ count = 0 # Iterate over the plugin to locate specific @plugin decorated methods for name in dir(plugin_module): plugin_method = getattr(plugin_module, name) plugin_opts = self._parse_plugin_opts(plugin_method) if plugin_opts is not None: plugin_impl = PluginImpl(plugin_module, plugin_method, plugin_opts) plugins_list = self._registry.setdefault(plugin_opts['extension_point_url'], []) plugins_list.append(plugin_impl) plugins_list.sort(key=lambda p: p.key) if self.DEBUG: print('Registered plugin with ID "{}" for extension point: {}'.format(plugin_impl.id, plugin_opts['extension_point_url'])) count += 1 return count
def _parse_plugin_opts(self, plugin_method): if not inspect.isroutine(plugin_method): return try: res = getattr(plugin_method, '__plugin_spec__', None) except Exception: res = {} if res is not None and not isinstance(res, dict): # false positive res = None return res
[docs]def pluggable(pluggable_method=None, category=None, selector='first_match', domain='https://plugins.compas.dev/'): """Decorator to mark a method as a pluggable extension point. A pluggable interface is uniquely identifiable/locatable via a URL derived from the parameters ``domain``, ``category`` and name ``pluggable_method``. In the example below, the URL of the pluggable definition is:: https://plugins.compas.dev/triangulation/triangulate_mesh Parameters ---------- pluggable_method : callable The method to decorate as ``pluggable``. category : :obj:`str`, optional An optional string to group or categorize extension points. selector : :obj:`str`, optional String that determines the selection mode of extension points. - ``"first_match"``: (:obj:`str`) Execute the first matching implementation. - ``"collect_all"``: (:obj:`str`) Executes all matching implementations and return list of its return values. domain : :obj:`str`, optional Domain name that "owns" the pluggable extension point. This is useful to avoid name collisions between extension points of different packages. Examples -------- >>> @pluggable(category='triangulation') ... def triangulate_mesh(mesh): ... pass """ def pluggable_decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): extension_point_url = _get_extension_point_url_from_method(domain, category, func) # Select first matching plugin if selector == 'first_match': plugin_impl = _select_plugin(extension_point_url) # Invoke plugin return plugin_impl.method(*args, **kwargs) # Collect all matching plugins elif selector == 'collect_all': results = [] for plugin_impl in _collect_plugins(extension_point_url): results.append(plugin_impl.method(*args, **kwargs)) return results else: raise ValueError('Unexpected selector type. Must be either: first_match or collect_all') return wrapper if pluggable_method is None: return pluggable_decorator else: return pluggable_decorator(pluggable_method)
[docs]def plugin(method=None, category=None, requires=None, tryfirst=False, trylast=False, pluggable_name=None, domain='https://plugins.compas.dev/'): """Decorator to declare a plugin. A plugin decorator marks a method as a plugin for a specified :meth:`pluggable` extension point. Plugins are matched to their pluggable counterparts by a combination of the name of the plugin method, the category and the domain specified. These 3 parts form the **extension point URL** used for matching. Parameters ---------- method : callable The method to decorate as ``plugin``. category : str, optional An optional string to group or categorize plugins. requires : list of str, optional Optionally defines a list of packages that should be importable for this plugin to be used. tryfirst : bool, optional Plugins can declare a preferred priority by setting this to ``True``. By default ``False``. trylast : bool, optional Alternatively, a plugin can demote itself to be least preferable setting ``trylast`` to ``True``. By default ``False``. pluggable_name : str, optional Usually, the name of the decorated plugin method matches that of the pluggable interface. When that is not the case, the pluggable name can be specified via this parameter. domain : str, optional Domain name that "owns" the pluggable extension point. This is useful to disambiguate name collisions between extension points of different packages. """ def setattr_hookspec_opts(func): if tryfirst and trylast: raise ValueError('You cannot set a plugin to try first and last at the same time.') name = pluggable_name or getattr(func, '__name__', None) extension_point_url = _get_extension_point_url_from_name(domain, category, name) setattr( func, '__plugin_spec__', dict( extension_point_url=extension_point_url, pluggable_name=name, requires=requires, tryfirst=tryfirst, trylast=trylast, ), ) return func if method is not None: return setattr_hookspec_opts(method) else: return setattr_hookspec_opts
class Importer(object): """Internal class to help importing modules.""" def __init__(self): # dictionary of module_name => bool (importable yes/no) self._cache = {} def try_import(self, module_name): """Attempt to import a module, but do not raise in case of error. Parameters ---------- module_name : str Module to try to import. Returns ------- module If importable, it returns the imported module, otherwise ``None``. """ module = None try: module = __import__(module_name, fromlist=['__name__'], level=0) self._cache[module_name] = True except ImportError: self._cache[module_name] = False return module def check_importable(self, module_name): """Check if a module is importable. Parameters ---------- module_name : str Name of the module to check for importability. Returns ------- bool ``True`` if the module can be imported correctly, otherwise ``False``. """ if module_name not in self._cache: self.try_import(module_name) return self._cache[module_name] def is_plugin_selectable(plugin, manager): if plugin.opts['requires']: importable_requirements = (manager.importer.check_importable(name) for name in plugin.opts['requires']) if not all(importable_requirements): if manager.DEBUG: print('Requirements not satisfied. Plugin will not be used: {}'.format(plugin.id)) return False return True def select_plugin(extension_point_url, manager): if manager.DEBUG: print('Extension Point URL {} invoked. Will select a matching plugin'.format(extension_point_url)) plugins = manager.registry.get(extension_point_url) or [] for plugin in plugins: if is_plugin_selectable(plugin, manager): return plugin # Nothing found, raise raise PluginNotInstalledError('Plugin not found for extension point URL: {}'.format(extension_point_url)) def collect_plugins(extension_point_url, manager): if manager.DEBUG: print('Extension Point URL {} invoked. Will select a matching plugin'.format(extension_point_url)) plugins = manager.registry.get(extension_point_url) or [] return [plugin for plugin in plugins if is_plugin_selectable(plugin, manager)] plugin_manager = PluginManager() _select_plugin = functools.partial(select_plugin, manager=plugin_manager) _collect_plugins = functools.partial(collect_plugins, manager=plugin_manager)