Mercurial > repos > shellac > sam_consensus_v3
diff env/lib/python3.9/site-packages/pluggy/hooks.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.9/site-packages/pluggy/hooks.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,359 @@ +""" +Internal hook annotation, representation and calling machinery. +""" +import inspect +import sys +import warnings +from .callers import _legacymulticall, _multicall + + +class HookspecMarker(object): + """ Decorator helper class for marking functions as hook specifications. + + You can instantiate it with a project_name to get a decorator. + Calling :py:meth:`.PluginManager.add_hookspecs` later will discover all marked functions + if the :py:class:`.PluginManager` uses the same project_name. + """ + + def __init__(self, project_name): + self.project_name = project_name + + def __call__( + self, function=None, firstresult=False, historic=False, warn_on_impl=None + ): + """ if passed a function, directly sets attributes on the function + which will make it discoverable to :py:meth:`.PluginManager.add_hookspecs`. + If passed no function, returns a decorator which can be applied to a function + later using the attributes supplied. + + If ``firstresult`` is ``True`` the 1:N hook call (N being the number of registered + hook implementation functions) will stop at I<=N when the I'th function + returns a non-``None`` result. + + If ``historic`` is ``True`` calls to a hook will be memorized and replayed + on later registered plugins. + + """ + + def setattr_hookspec_opts(func): + if historic and firstresult: + raise ValueError("cannot have a historic firstresult hook") + setattr( + func, + self.project_name + "_spec", + dict( + firstresult=firstresult, + historic=historic, + warn_on_impl=warn_on_impl, + ), + ) + return func + + if function is not None: + return setattr_hookspec_opts(function) + else: + return setattr_hookspec_opts + + +class HookimplMarker(object): + """ Decorator helper class for marking functions as hook implementations. + + You can instantiate with a ``project_name`` to get a decorator. + Calling :py:meth:`.PluginManager.register` later will discover all marked functions + if the :py:class:`.PluginManager` uses the same project_name. + """ + + def __init__(self, project_name): + self.project_name = project_name + + def __call__( + self, + function=None, + hookwrapper=False, + optionalhook=False, + tryfirst=False, + trylast=False, + ): + + """ if passed a function, directly sets attributes on the function + which will make it discoverable to :py:meth:`.PluginManager.register`. + If passed no function, returns a decorator which can be applied to a + function later using the attributes supplied. + + If ``optionalhook`` is ``True`` a missing matching hook specification will not result + in an error (by default it is an error if no matching spec is found). + + If ``tryfirst`` is ``True`` this hook implementation will run as early as possible + in the chain of N hook implementations for a specification. + + If ``trylast`` is ``True`` this hook implementation will run as late as possible + in the chain of N hook implementations. + + If ``hookwrapper`` is ``True`` the hook implementations needs to execute exactly + one ``yield``. The code before the ``yield`` is run early before any non-hookwrapper + function is run. The code after the ``yield`` is run after all non-hookwrapper + function have run. The ``yield`` receives a :py:class:`.callers._Result` object + representing the exception or result outcome of the inner calls (including other + hookwrapper calls). + + """ + + def setattr_hookimpl_opts(func): + setattr( + func, + self.project_name + "_impl", + dict( + hookwrapper=hookwrapper, + optionalhook=optionalhook, + tryfirst=tryfirst, + trylast=trylast, + ), + ) + return func + + if function is None: + return setattr_hookimpl_opts + else: + return setattr_hookimpl_opts(function) + + +def normalize_hookimpl_opts(opts): + opts.setdefault("tryfirst", False) + opts.setdefault("trylast", False) + opts.setdefault("hookwrapper", False) + opts.setdefault("optionalhook", False) + + +if hasattr(inspect, "getfullargspec"): + + def _getargspec(func): + return inspect.getfullargspec(func) + + +else: + + def _getargspec(func): + return inspect.getargspec(func) + + +_PYPY3 = hasattr(sys, "pypy_version_info") and sys.version_info.major == 3 + + +def varnames(func): + """Return tuple of positional and keywrord argument names for a function, + method, class or callable. + + In case of a class, its ``__init__`` method is considered. + For methods the ``self`` parameter is not included. + """ + cache = getattr(func, "__dict__", {}) + try: + return cache["_varnames"] + except KeyError: + pass + + if inspect.isclass(func): + try: + func = func.__init__ + except AttributeError: + return (), () + elif not inspect.isroutine(func): # callable object? + try: + func = getattr(func, "__call__", func) + except Exception: + return (), () + + try: # func MUST be a function or method here or we won't parse any args + spec = _getargspec(func) + except TypeError: + return (), () + + args, defaults = tuple(spec.args), spec.defaults + if defaults: + index = -len(defaults) + args, kwargs = args[:index], tuple(args[index:]) + else: + kwargs = () + + # strip any implicit instance arg + # pypy3 uses "obj" instead of "self" for default dunder methods + implicit_names = ("self",) if not _PYPY3 else ("self", "obj") + if args: + if inspect.ismethod(func) or ( + "." in getattr(func, "__qualname__", ()) and args[0] in implicit_names + ): + args = args[1:] + + try: + cache["_varnames"] = args, kwargs + except TypeError: + pass + return args, kwargs + + +class _HookRelay(object): + """ hook holder object for performing 1:N hook calls where N is the number + of registered plugins. + + """ + + +class _HookCaller(object): + def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None): + self.name = name + self._wrappers = [] + self._nonwrappers = [] + self._hookexec = hook_execute + self.argnames = None + self.kwargnames = None + self.multicall = _multicall + self.spec = None + if specmodule_or_class is not None: + assert spec_opts is not None + self.set_specification(specmodule_or_class, spec_opts) + + def has_spec(self): + return self.spec is not None + + def set_specification(self, specmodule_or_class, spec_opts): + assert not self.has_spec() + self.spec = HookSpec(specmodule_or_class, self.name, spec_opts) + if spec_opts.get("historic"): + self._call_history = [] + + def is_historic(self): + return hasattr(self, "_call_history") + + def _remove_plugin(self, plugin): + def remove(wrappers): + for i, method in enumerate(wrappers): + if method.plugin == plugin: + del wrappers[i] + return True + + if remove(self._wrappers) is None: + if remove(self._nonwrappers) is None: + raise ValueError("plugin %r not found" % (plugin,)) + + def get_hookimpls(self): + # Order is important for _hookexec + return self._nonwrappers + self._wrappers + + def _add_hookimpl(self, hookimpl): + """Add an implementation to the callback chain. + """ + if hookimpl.hookwrapper: + methods = self._wrappers + else: + methods = self._nonwrappers + + if hookimpl.trylast: + methods.insert(0, hookimpl) + elif hookimpl.tryfirst: + methods.append(hookimpl) + else: + # find last non-tryfirst method + i = len(methods) - 1 + while i >= 0 and methods[i].tryfirst: + i -= 1 + methods.insert(i + 1, hookimpl) + + if "__multicall__" in hookimpl.argnames: + warnings.warn( + "Support for __multicall__ is now deprecated and will be" + "removed in an upcoming release.", + DeprecationWarning, + ) + self.multicall = _legacymulticall + + def __repr__(self): + return "<_HookCaller %r>" % (self.name,) + + def __call__(self, *args, **kwargs): + if args: + raise TypeError("hook calling supports only keyword arguments") + assert not self.is_historic() + if self.spec and self.spec.argnames: + notincall = ( + set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys()) + ) + if notincall: + warnings.warn( + "Argument(s) {} which are declared in the hookspec " + "can not be found in this hook call".format(tuple(notincall)), + stacklevel=2, + ) + return self._hookexec(self, self.get_hookimpls(), kwargs) + + def call_historic(self, result_callback=None, kwargs=None, proc=None): + """Call the hook with given ``kwargs`` for all registered plugins and + for all plugins which will be registered afterwards. + + If ``result_callback`` is not ``None`` it will be called for for each + non-``None`` result obtained from a hook implementation. + + .. note:: + The ``proc`` argument is now deprecated. + """ + if proc is not None: + warnings.warn( + "Support for `proc` argument is now deprecated and will be" + "removed in an upcoming release.", + DeprecationWarning, + ) + result_callback = proc + + self._call_history.append((kwargs or {}, result_callback)) + # historizing hooks don't return results + res = self._hookexec(self, self.get_hookimpls(), kwargs) + if result_callback is None: + return + # XXX: remember firstresult isn't compat with historic + for x in res or []: + result_callback(x) + + def call_extra(self, methods, kwargs): + """ Call the hook with some additional temporarily participating + methods using the specified ``kwargs`` as call parameters. """ + old = list(self._nonwrappers), list(self._wrappers) + for method in methods: + opts = dict(hookwrapper=False, trylast=False, tryfirst=False) + hookimpl = HookImpl(None, "<temp>", method, opts) + self._add_hookimpl(hookimpl) + try: + return self(**kwargs) + finally: + self._nonwrappers, self._wrappers = old + + def _maybe_apply_history(self, method): + """Apply call history to a new hookimpl if it is marked as historic. + """ + if self.is_historic(): + for kwargs, result_callback in self._call_history: + res = self._hookexec(self, [method], kwargs) + if res and result_callback is not None: + result_callback(res[0]) + + +class HookImpl(object): + def __init__(self, plugin, plugin_name, function, hook_impl_opts): + self.function = function + self.argnames, self.kwargnames = varnames(self.function) + self.plugin = plugin + self.opts = hook_impl_opts + self.plugin_name = plugin_name + self.__dict__.update(hook_impl_opts) + + def __repr__(self): + return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin) + + +class HookSpec(object): + def __init__(self, namespace, name, opts): + self.namespace = namespace + self.function = function = getattr(namespace, name) + self.name = name + self.argnames, self.kwargnames = varnames(function) + self.opts = opts + self.argnames = ["__multicall__"] + list(self.argnames) + self.warn_on_impl = opts.get("warn_on_impl")