123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778 |
- """
- PluginManager, basic initialization and tracing.
- pluggy is the cristallized core of plugin management as used
- by some 150 plugins for pytest.
- Pluggy uses semantic versioning. Breaking changes are only foreseen for
- Major releases (incremented X in "X.Y.Z"). If you want to use pluggy in
- your project you should thus use a dependency restriction like
- "pluggy>=0.1.0,<1.0" to avoid surprises.
- pluggy is concerned with hook specification, hook implementations and hook
- calling. For any given hook specification a hook call invokes up to N implementations.
- A hook implementation can influence its position and type of execution:
- if attributed "tryfirst" or "trylast" it will be tried to execute
- first or last. However, if attributed "hookwrapper" an implementation
- can wrap all calls to non-hookwrapper implementations. A hookwrapper
- can thus execute some code ahead and after the execution of other hooks.
- Hook specification is done by way of a regular python function where
- both the function name and the names of all its arguments are significant.
- Each hook implementation function is verified against the original specification
- function, including the names of all its arguments. To allow for hook specifications
- to evolve over the livetime of a project, hook implementations can
- accept less arguments. One can thus add new arguments and semantics to
- a hook specification by adding another argument typically without breaking
- existing hook implementations.
- The chosen approach is meant to let a hook designer think carefuly about
- which objects are needed by an extension writer. By contrast, subclass-based
- extension mechanisms often expose a lot more state and behaviour than needed,
- thus restricting future developments.
- Pluggy currently consists of functionality for:
- - a way to register new hook specifications. Without a hook
- specification no hook calling can be performed.
- - a registry of plugins which contain hook implementation functions. It
- is possible to register plugins for which a hook specification is not yet
- known and validate all hooks when the system is in a more referentially
- consistent state. Setting an "optionalhook" attribution to a hook
- implementation will avoid PluginValidationError's if a specification
- is missing. This allows to have optional integration between plugins.
- - a "hook" relay object from which you can launch 1:N calls to
- registered hook implementation functions
- - a mechanism for ordering hook implementation functions
- - mechanisms for two different type of 1:N calls: "firstresult" for when
- the call should stop when the first implementation returns a non-None result.
- And the other (default) way of guaranteeing that all hook implementations
- will be called and their non-None result collected.
- - mechanisms for "historic" extension points such that all newly
- registered functions will receive all hook calls that happened
- before their registration.
- - a mechanism for discovering plugin objects which are based on
- setuptools based entry points.
- - a simple tracing mechanism, including tracing of plugin calls and
- their arguments.
- """
- import sys
- import inspect
- __version__ = '0.3.1'
- __all__ = ["PluginManager", "PluginValidationError",
- "HookspecMarker", "HookimplMarker"]
- _py3 = sys.version_info > (3, 0)
- class HookspecMarker:
- """ Decorator helper class for marking functions as hook specifications.
- You can instantiate it with a project_name to get a decorator.
- Calling PluginManager.add_hookspecs later will discover all marked functions
- if the 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):
- """ if passed a function, directly sets attributes on the function
- which will make it discoverable to 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))
- return func
- if function is not None:
- return setattr_hookspec_opts(function)
- else:
- return setattr_hookspec_opts
- class HookimplMarker:
- """ Decorator helper class for marking functions as hook implementations.
- You can instantiate with a project_name to get a decorator.
- Calling PluginManager.register later will discover all marked functions
- if the 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 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 specfication.
- 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 an ``_CallOutcome`` 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)
- class _TagTracer:
- def __init__(self):
- self._tag2proc = {}
- self.writer = None
- self.indent = 0
- def get(self, name):
- return _TagTracerSub(self, (name,))
- def format_message(self, tags, args):
- if isinstance(args[-1], dict):
- extra = args[-1]
- args = args[:-1]
- else:
- extra = {}
- content = " ".join(map(str, args))
- indent = " " * self.indent
- lines = [
- "%s%s [%s]\n" % (indent, content, ":".join(tags))
- ]
- for name, value in extra.items():
- lines.append("%s %s: %s\n" % (indent, name, value))
- return lines
- def processmessage(self, tags, args):
- if self.writer is not None and args:
- lines = self.format_message(tags, args)
- self.writer(''.join(lines))
- try:
- self._tag2proc[tags](tags, args)
- except KeyError:
- pass
- def setwriter(self, writer):
- self.writer = writer
- def setprocessor(self, tags, processor):
- if isinstance(tags, str):
- tags = tuple(tags.split(":"))
- else:
- assert isinstance(tags, tuple)
- self._tag2proc[tags] = processor
- class _TagTracerSub:
- def __init__(self, root, tags):
- self.root = root
- self.tags = tags
- def __call__(self, *args):
- self.root.processmessage(self.tags, args)
- def setmyprocessor(self, processor):
- self.root.setprocessor(self.tags, processor)
- def get(self, name):
- return self.__class__(self.root, self.tags + (name,))
- def _raise_wrapfail(wrap_controller, msg):
- co = wrap_controller.gi_code
- raise RuntimeError("wrap_controller at %r %s:%d %s" %
- (co.co_name, co.co_filename, co.co_firstlineno, msg))
- def _wrapped_call(wrap_controller, func):
- """ Wrap calling to a function with a generator which needs to yield
- exactly once. The yield point will trigger calling the wrapped function
- and return its _CallOutcome to the yield point. The generator then needs
- to finish (raise StopIteration) in order for the wrapped call to complete.
- """
- try:
- next(wrap_controller) # first yield
- except StopIteration:
- _raise_wrapfail(wrap_controller, "did not yield")
- call_outcome = _CallOutcome(func)
- try:
- wrap_controller.send(call_outcome)
- _raise_wrapfail(wrap_controller, "has second yield")
- except StopIteration:
- pass
- return call_outcome.get_result()
- class _CallOutcome:
- """ Outcome of a function call, either an exception or a proper result.
- Calling the ``get_result`` method will return the result or reraise
- the exception raised when the function was called. """
- excinfo = None
- def __init__(self, func):
- try:
- self.result = func()
- except BaseException:
- self.excinfo = sys.exc_info()
- def force_result(self, result):
- self.result = result
- self.excinfo = None
- def get_result(self):
- if self.excinfo is None:
- return self.result
- else:
- ex = self.excinfo
- if _py3:
- raise ex[1].with_traceback(ex[2])
- _reraise(*ex) # noqa
- if not _py3:
- exec("""
- def _reraise(cls, val, tb):
- raise cls, val, tb
- """)
- class _TracedHookExecution:
- def __init__(self, pluginmanager, before, after):
- self.pluginmanager = pluginmanager
- self.before = before
- self.after = after
- self.oldcall = pluginmanager._inner_hookexec
- assert not isinstance(self.oldcall, _TracedHookExecution)
- self.pluginmanager._inner_hookexec = self
- def __call__(self, hook, hook_impls, kwargs):
- self.before(hook.name, hook_impls, kwargs)
- outcome = _CallOutcome(lambda: self.oldcall(hook, hook_impls, kwargs))
- self.after(outcome, hook.name, hook_impls, kwargs)
- return outcome.get_result()
- def undo(self):
- self.pluginmanager._inner_hookexec = self.oldcall
- class PluginManager(object):
- """ Core Pluginmanager class which manages registration
- of plugin objects and 1:N hook calling.
- You can register new hooks by calling ``addhooks(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. An optional excludefunc allows to blacklist names which
- are not considered as hooks despite a matching prefix.
- 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 = _TagTracer().get("pluginmanage")
- self.hook = _HookRelay(self.trace.root.get("hook"))
- self._implprefix = implprefix
- self._inner_hookexec = lambda hook, methods, kwargs: \
- _MultiCall(methods, kwargs, hook.spec_opts).execute()
- 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)
- res = getattr(method, self.project_name + "_impl", None)
- if res is not None and not isinstance(res, dict):
- # false positive
- res = None
- elif res is None and self._implprefix and name.startswith(self._implprefix):
- 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 name blogs registering plugins of that name. """
- 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._wrappers + hc._nonwrappers):
- 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 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(
- "Plugin %r\nhook %r\nhistoric incompatible to hookwrapper" %
- (hookimpl.plugin_name, hook.name))
- for arg in hookimpl.argnames:
- if arg not in hook.argnames:
- raise PluginValidationError(
- "Plugin %r\nhook %r\nargument %r not available\n"
- "plugin definition: %s\n"
- "available hookargs: %s" %
- (hookimpl.plugin_name, hook.name, arg,
- _formatdef(hookimpl.function), ", ".join(hook.argnames)))
- 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._wrappers + hook._nonwrappers):
- if not hookimpl.optionalhook:
- raise PluginValidationError(
- "unknown hook %r in plugin %r" %
- (name, hookimpl.plugin))
- def load_setuptools_entrypoints(self, entrypoint_name):
- """ Load modules from querying the specified setuptools entrypoint name.
- Return the number of loaded plugins. """
- from pkg_resources import iter_entry_points, DistributionNotFound
- for ep in iter_entry_points(entrypoint_name):
- # is the plugin registered or blocked?
- if self.get_plugin(ep.name) or self.is_blocked(ep.name):
- continue
- try:
- plugin = ep.load()
- except DistributionNotFound:
- continue
- self.register(plugin, name=ep.name)
- self._plugin_distinfo.append((plugin, ep.dist))
- return len(self._plugin_distinfo)
- 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:`_CallOutcome`` object
- which represents the result of the overall hook call.
- """
- return _TracedHookExecution(self, before, after).undo
- def enable_tracing(self):
- """ enable tracing of hook calls and return an undo function. """
- hooktrace = self.hook._trace
- 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.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._specmodule_or_class,
- orig.spec_opts)
- for hookimpl in (orig._wrappers + orig._nonwrappers):
- 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
- class _MultiCall:
- """ execute a call into multiple python functions/methods. """
- # XXX note that the __multicall__ argument is supported only
- # for pytest compatibility reasons. It was never officially
- # supported there and is explicitly deprecated since 2.8
- # so we can remove it soon, allowing to avoid the below recursion
- # in execute() and simplify/speed up the execute loop.
- def __init__(self, hook_impls, kwargs, specopts={}):
- self.hook_impls = hook_impls
- self.kwargs = kwargs
- self.kwargs["__multicall__"] = self
- self.specopts = specopts
- def execute(self):
- all_kwargs = self.kwargs
- self.results = results = []
- firstresult = self.specopts.get("firstresult")
- while self.hook_impls:
- hook_impl = self.hook_impls.pop()
- args = [all_kwargs[argname] for argname in hook_impl.argnames]
- if hook_impl.hookwrapper:
- return _wrapped_call(hook_impl.function(*args), self.execute)
- res = hook_impl.function(*args)
- if res is not None:
- if firstresult:
- return res
- results.append(res)
- if not firstresult:
- return results
- def __repr__(self):
- status = "%d meths" % (len(self.hook_impls),)
- if hasattr(self, "results"):
- status = ("%d results, " % len(self.results)) + status
- return "<_MultiCall %s, kwargs=%r>" % (status, self.kwargs)
- def varnames(func, startindex=None):
- """ return argument name tuple 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 unless you are passing
- an unbound method with Python3 (which has no supports for unbound methods)
- """
- cache = getattr(func, "__dict__", {})
- try:
- return cache["_varnames"]
- except KeyError:
- pass
- if inspect.isclass(func):
- try:
- func = func.__init__
- except AttributeError:
- return ()
- startindex = 1
- else:
- if not inspect.isfunction(func) and not inspect.ismethod(func):
- func = getattr(func, '__call__', func)
- if startindex is None:
- startindex = int(inspect.ismethod(func))
- try:
- rawcode = func.__code__
- except AttributeError:
- return ()
- try:
- x = rawcode.co_varnames[startindex:rawcode.co_argcount]
- except AttributeError:
- x = ()
- else:
- defaults = func.__defaults__
- if defaults:
- x = x[:-len(defaults)]
- try:
- cache["_varnames"] = x
- except TypeError:
- pass
- return x
- class _HookRelay:
- """ hook holder object for performing 1:N hook calls where N is the number
- of registered plugins.
- """
- def __init__(self, trace):
- self._trace = trace
- 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
- 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 hasattr(self, "_specmodule_or_class")
- def set_specification(self, specmodule_or_class, spec_opts):
- assert not self.has_spec()
- self._specmodule_or_class = specmodule_or_class
- specfunc = getattr(specmodule_or_class, self.name)
- argnames = varnames(specfunc, startindex=inspect.isclass(specmodule_or_class))
- assert "self" not in argnames # sanity check
- self.argnames = ["__multicall__"] + list(argnames)
- self.spec_opts = 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 _add_hookimpl(self, hookimpl):
- 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)
- def __repr__(self):
- return "<_HookCaller %r>" % (self.name,)
- def __call__(self, **kwargs):
- assert not self.is_historic()
- return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
- def call_historic(self, proc=None, kwargs=None):
- self._call_history.append((kwargs or {}, proc))
- # historizing hooks don't return results
- self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
- 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):
- if self.is_historic():
- for kwargs, proc in self._call_history:
- res = self._hookexec(self, [method], kwargs)
- if res and proc is not None:
- proc(res[0])
- class HookImpl:
- def __init__(self, plugin, plugin_name, function, hook_impl_opts):
- self.function = function
- self.argnames = varnames(self.function)
- self.plugin = plugin
- self.opts = hook_impl_opts
- self.plugin_name = plugin_name
- self.__dict__.update(hook_impl_opts)
- class PluginValidationError(Exception):
- """ plugin failed validation. """
- 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))
- )
|