|
|
- 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 :py:class:`.PluginManager` class which manages registration
- of plugin objects and 1:N hook calling.
-
- You can register new hooks by calling :py:meth:`add_hookspecs(module_or_class)
- <.PluginManager.add_hookspecs>`.
- You can register plugin objects (which contain hooks) by calling
- :py:meth:`register(plugin) <.PluginManager.register>`. The :py:class:`.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 :py:meth:`.PluginManager.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 :py:class:`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 :py:meth:`register(plugin, name) <.PluginManager.register>`.
- To obtain the name of an registered plugin use :py:meth:`get_name(plugin)
- <.PluginManager.get_name>` 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 :py:class:`.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:`pluggy.callers._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 :py:class:`.hooks._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)),
- )
|