392 lines
15 KiB
Python
392 lines
15 KiB
Python
import inspect
|
|
import sys
|
|
from . import _tracing
|
|
from .callers import _Result
|
|
from .hooks import HookImpl, _HookRelay, _HookCaller, normalize_hookimpl_opts
|
|
import warnings
|
|
|
|
if sys.version_info >= (3, 8):
|
|
from importlib import metadata as importlib_metadata
|
|
else:
|
|
import importlib_metadata
|
|
|
|
|
|
def _warn_for_function(warning, function):
|
|
warnings.warn_explicit(
|
|
warning,
|
|
type(warning),
|
|
lineno=function.__code__.co_firstlineno,
|
|
filename=function.__code__.co_filename,
|
|
)
|
|
|
|
|
|
class PluginValidationError(Exception):
|
|
""" plugin failed validation.
|
|
|
|
:param object plugin: the plugin which failed validation,
|
|
may be a module or an arbitrary object.
|
|
"""
|
|
|
|
def __init__(self, plugin, message):
|
|
self.plugin = plugin
|
|
super(Exception, self).__init__(message)
|
|
|
|
|
|
class DistFacade(object):
|
|
"""Emulate a pkg_resources Distribution"""
|
|
|
|
def __init__(self, dist):
|
|
self._dist = dist
|
|
|
|
@property
|
|
def project_name(self):
|
|
return self.metadata["name"]
|
|
|
|
def __getattr__(self, attr, default=None):
|
|
return getattr(self._dist, attr, default)
|
|
|
|
def __dir__(self):
|
|
return sorted(dir(self._dist) + ["_dist", "project_name"])
|
|
|
|
|
|
class PluginManager(object):
|
|
""" Core Pluginmanager class which manages registration
|
|
of plugin objects and 1:N hook calling.
|
|
|
|
You can register new hooks by calling ``add_hookspecs(module_or_class)``.
|
|
You can register plugin objects (which contain hooks) by calling
|
|
``register(plugin)``. The Pluginmanager is initialized with a
|
|
prefix that is searched for in the names of the dict of registered
|
|
plugin objects.
|
|
|
|
For debugging purposes you can call ``enable_tracing()``
|
|
which will subsequently send debug information to the trace helper.
|
|
"""
|
|
|
|
def __init__(self, project_name, implprefix=None):
|
|
"""If ``implprefix`` is given implementation functions
|
|
will be recognized if their name matches the implprefix. """
|
|
self.project_name = project_name
|
|
self._name2plugin = {}
|
|
self._plugin2hookcallers = {}
|
|
self._plugin_distinfo = []
|
|
self.trace = _tracing.TagTracer().get("pluginmanage")
|
|
self.hook = _HookRelay()
|
|
if implprefix is not None:
|
|
warnings.warn(
|
|
"Support for the `implprefix` arg is now deprecated and will "
|
|
"be removed in an upcoming release. Please use HookimplMarker.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
self._implprefix = implprefix
|
|
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
|
|
methods,
|
|
kwargs,
|
|
firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
|
|
)
|
|
|
|
def _hookexec(self, hook, methods, kwargs):
|
|
# called from all hookcaller instances.
|
|
# enable_tracing will set its own wrapping function at self._inner_hookexec
|
|
return self._inner_hookexec(hook, methods, kwargs)
|
|
|
|
def register(self, plugin, name=None):
|
|
""" Register a plugin and return its canonical name or None if the name
|
|
is blocked from registering. Raise a ValueError if the plugin is already
|
|
registered. """
|
|
plugin_name = name or self.get_canonical_name(plugin)
|
|
|
|
if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
|
|
if self._name2plugin.get(plugin_name, -1) is None:
|
|
return # blocked plugin, return None to indicate no registration
|
|
raise ValueError(
|
|
"Plugin already registered: %s=%s\n%s"
|
|
% (plugin_name, plugin, self._name2plugin)
|
|
)
|
|
|
|
# XXX if an error happens we should make sure no state has been
|
|
# changed at point of return
|
|
self._name2plugin[plugin_name] = plugin
|
|
|
|
# register matching hook implementations of the plugin
|
|
self._plugin2hookcallers[plugin] = hookcallers = []
|
|
for name in dir(plugin):
|
|
hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
|
|
if hookimpl_opts is not None:
|
|
normalize_hookimpl_opts(hookimpl_opts)
|
|
method = getattr(plugin, name)
|
|
hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
|
|
hook = getattr(self.hook, name, None)
|
|
if hook is None:
|
|
hook = _HookCaller(name, self._hookexec)
|
|
setattr(self.hook, name, hook)
|
|
elif hook.has_spec():
|
|
self._verify_hook(hook, hookimpl)
|
|
hook._maybe_apply_history(hookimpl)
|
|
hook._add_hookimpl(hookimpl)
|
|
hookcallers.append(hook)
|
|
return plugin_name
|
|
|
|
def parse_hookimpl_opts(self, plugin, name):
|
|
method = getattr(plugin, name)
|
|
if not inspect.isroutine(method):
|
|
return
|
|
try:
|
|
res = getattr(method, self.project_name + "_impl", None)
|
|
except Exception:
|
|
res = {}
|
|
if res is not None and not isinstance(res, dict):
|
|
# false positive
|
|
res = None
|
|
# TODO: remove when we drop implprefix in 1.0
|
|
elif res is None and self._implprefix and name.startswith(self._implprefix):
|
|
_warn_for_function(
|
|
DeprecationWarning(
|
|
"The `implprefix` system is deprecated please decorate "
|
|
"this function using an instance of HookimplMarker."
|
|
),
|
|
method,
|
|
)
|
|
res = {}
|
|
return res
|
|
|
|
def unregister(self, plugin=None, name=None):
|
|
""" unregister a plugin object and all its contained hook implementations
|
|
from internal data structures. """
|
|
if name is None:
|
|
assert plugin is not None, "one of name or plugin needs to be specified"
|
|
name = self.get_name(plugin)
|
|
|
|
if plugin is None:
|
|
plugin = self.get_plugin(name)
|
|
|
|
# if self._name2plugin[name] == None registration was blocked: ignore
|
|
if self._name2plugin.get(name):
|
|
del self._name2plugin[name]
|
|
|
|
for hookcaller in self._plugin2hookcallers.pop(plugin, []):
|
|
hookcaller._remove_plugin(plugin)
|
|
|
|
return plugin
|
|
|
|
def set_blocked(self, name):
|
|
""" block registrations of the given name, unregister if already registered. """
|
|
self.unregister(name=name)
|
|
self._name2plugin[name] = None
|
|
|
|
def is_blocked(self, name):
|
|
""" return True if the given plugin name is blocked. """
|
|
return name in self._name2plugin and self._name2plugin[name] is None
|
|
|
|
def add_hookspecs(self, module_or_class):
|
|
""" add new hook specifications defined in the given module_or_class.
|
|
Functions are recognized if they have been decorated accordingly. """
|
|
names = []
|
|
for name in dir(module_or_class):
|
|
spec_opts = self.parse_hookspec_opts(module_or_class, name)
|
|
if spec_opts is not None:
|
|
hc = getattr(self.hook, name, None)
|
|
if hc is None:
|
|
hc = _HookCaller(name, self._hookexec, module_or_class, spec_opts)
|
|
setattr(self.hook, name, hc)
|
|
else:
|
|
# plugins registered this hook without knowing the spec
|
|
hc.set_specification(module_or_class, spec_opts)
|
|
for hookfunction in hc.get_hookimpls():
|
|
self._verify_hook(hc, hookfunction)
|
|
names.append(name)
|
|
|
|
if not names:
|
|
raise ValueError(
|
|
"did not find any %r hooks in %r" % (self.project_name, module_or_class)
|
|
)
|
|
|
|
def parse_hookspec_opts(self, module_or_class, name):
|
|
method = getattr(module_or_class, name)
|
|
return getattr(method, self.project_name + "_spec", None)
|
|
|
|
def get_plugins(self):
|
|
""" return the set of registered plugins. """
|
|
return set(self._plugin2hookcallers)
|
|
|
|
def is_registered(self, plugin):
|
|
""" Return True if the plugin is already registered. """
|
|
return plugin in self._plugin2hookcallers
|
|
|
|
def get_canonical_name(self, plugin):
|
|
""" Return canonical name for a plugin object. Note that a plugin
|
|
may be registered under a different name which was specified
|
|
by the caller of register(plugin, name). To obtain the name
|
|
of an registered plugin use ``get_name(plugin)`` instead."""
|
|
return getattr(plugin, "__name__", None) or str(id(plugin))
|
|
|
|
def get_plugin(self, name):
|
|
""" Return a plugin or None for the given name. """
|
|
return self._name2plugin.get(name)
|
|
|
|
def has_plugin(self, name):
|
|
""" Return True if a plugin with the given name is registered. """
|
|
return self.get_plugin(name) is not None
|
|
|
|
def get_name(self, plugin):
|
|
""" Return name for registered plugin or None if not registered. """
|
|
for name, val in self._name2plugin.items():
|
|
if plugin == val:
|
|
return name
|
|
|
|
def _verify_hook(self, hook, hookimpl):
|
|
if hook.is_historic() and hookimpl.hookwrapper:
|
|
raise PluginValidationError(
|
|
hookimpl.plugin,
|
|
"Plugin %r\nhook %r\nhistoric incompatible to hookwrapper"
|
|
% (hookimpl.plugin_name, hook.name),
|
|
)
|
|
if hook.spec.warn_on_impl:
|
|
_warn_for_function(hook.spec.warn_on_impl, hookimpl.function)
|
|
# positional arg checking
|
|
notinspec = set(hookimpl.argnames) - set(hook.spec.argnames)
|
|
if notinspec:
|
|
raise PluginValidationError(
|
|
hookimpl.plugin,
|
|
"Plugin %r for hook %r\nhookimpl definition: %s\n"
|
|
"Argument(s) %s are declared in the hookimpl but "
|
|
"can not be found in the hookspec"
|
|
% (
|
|
hookimpl.plugin_name,
|
|
hook.name,
|
|
_formatdef(hookimpl.function),
|
|
notinspec,
|
|
),
|
|
)
|
|
|
|
def check_pending(self):
|
|
""" Verify that all hooks which have not been verified against
|
|
a hook specification are optional, otherwise raise PluginValidationError"""
|
|
for name in self.hook.__dict__:
|
|
if name[0] != "_":
|
|
hook = getattr(self.hook, name)
|
|
if not hook.has_spec():
|
|
for hookimpl in hook.get_hookimpls():
|
|
if not hookimpl.optionalhook:
|
|
raise PluginValidationError(
|
|
hookimpl.plugin,
|
|
"unknown hook %r in plugin %r"
|
|
% (name, hookimpl.plugin),
|
|
)
|
|
|
|
def load_setuptools_entrypoints(self, group, name=None):
|
|
""" Load modules from querying the specified setuptools ``group``.
|
|
|
|
:param str group: entry point group to load plugins
|
|
:param str name: if given, loads only plugins with the given ``name``.
|
|
:rtype: int
|
|
:return: return the number of loaded plugins by this call.
|
|
"""
|
|
count = 0
|
|
for dist in importlib_metadata.distributions():
|
|
for ep in dist.entry_points:
|
|
if (
|
|
ep.group != group
|
|
or (name is not None and ep.name != name)
|
|
# already registered
|
|
or self.get_plugin(ep.name)
|
|
or self.is_blocked(ep.name)
|
|
):
|
|
continue
|
|
plugin = ep.load()
|
|
self.register(plugin, name=ep.name)
|
|
self._plugin_distinfo.append((plugin, DistFacade(dist)))
|
|
count += 1
|
|
return count
|
|
|
|
def list_plugin_distinfo(self):
|
|
""" return list of distinfo/plugin tuples for all setuptools registered
|
|
plugins. """
|
|
return list(self._plugin_distinfo)
|
|
|
|
def list_name_plugin(self):
|
|
""" return list of name/plugin pairs. """
|
|
return list(self._name2plugin.items())
|
|
|
|
def get_hookcallers(self, plugin):
|
|
""" get all hook callers for the specified plugin. """
|
|
return self._plugin2hookcallers.get(plugin)
|
|
|
|
def add_hookcall_monitoring(self, before, after):
|
|
""" add before/after tracing functions for all hooks
|
|
and return an undo function which, when called,
|
|
will remove the added tracers.
|
|
|
|
``before(hook_name, hook_impls, kwargs)`` will be called ahead
|
|
of all hook calls and receive a hookcaller instance, a list
|
|
of HookImpl instances and the keyword arguments for the hook call.
|
|
|
|
``after(outcome, hook_name, hook_impls, kwargs)`` receives the
|
|
same arguments as ``before`` but also a :py:class:`_Result`` object
|
|
which represents the result of the overall hook call.
|
|
"""
|
|
oldcall = self._inner_hookexec
|
|
|
|
def traced_hookexec(hook, hook_impls, kwargs):
|
|
before(hook.name, hook_impls, kwargs)
|
|
outcome = _Result.from_call(lambda: oldcall(hook, hook_impls, kwargs))
|
|
after(outcome, hook.name, hook_impls, kwargs)
|
|
return outcome.get_result()
|
|
|
|
self._inner_hookexec = traced_hookexec
|
|
|
|
def undo():
|
|
self._inner_hookexec = oldcall
|
|
|
|
return undo
|
|
|
|
def enable_tracing(self):
|
|
""" enable tracing of hook calls and return an undo function. """
|
|
hooktrace = self.trace.root.get("hook")
|
|
|
|
def before(hook_name, methods, kwargs):
|
|
hooktrace.root.indent += 1
|
|
hooktrace(hook_name, kwargs)
|
|
|
|
def after(outcome, hook_name, methods, kwargs):
|
|
if outcome.excinfo is None:
|
|
hooktrace("finish", hook_name, "-->", outcome.get_result())
|
|
hooktrace.root.indent -= 1
|
|
|
|
return self.add_hookcall_monitoring(before, after)
|
|
|
|
def subset_hook_caller(self, name, remove_plugins):
|
|
""" Return a new _HookCaller instance for the named method
|
|
which manages calls to all registered plugins except the
|
|
ones from remove_plugins. """
|
|
orig = getattr(self.hook, name)
|
|
plugins_to_remove = [plug for plug in remove_plugins if hasattr(plug, name)]
|
|
if plugins_to_remove:
|
|
hc = _HookCaller(
|
|
orig.name, orig._hookexec, orig.spec.namespace, orig.spec.opts
|
|
)
|
|
for hookimpl in orig.get_hookimpls():
|
|
plugin = hookimpl.plugin
|
|
if plugin not in plugins_to_remove:
|
|
hc._add_hookimpl(hookimpl)
|
|
# we also keep track of this hook caller so it
|
|
# gets properly removed on plugin unregistration
|
|
self._plugin2hookcallers.setdefault(plugin, []).append(hc)
|
|
return hc
|
|
return orig
|
|
|
|
|
|
if hasattr(inspect, "signature"):
|
|
|
|
def _formatdef(func):
|
|
return "%s%s" % (func.__name__, str(inspect.signature(func)))
|
|
|
|
|
|
else:
|
|
|
|
def _formatdef(func):
|
|
return "%s%s" % (
|
|
func.__name__,
|
|
inspect.formatargspec(*inspect.getargspec(func)),
|
|
)
|