Coverage for /usr/local/lib/python3.7/site-packages/pluggy/hooks.py : 42%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Internal hook annotation, representation and calling machinery.
3"""
4import inspect
5import sys
6import warnings
7from .callers import _legacymulticall, _multicall
10class HookspecMarker(object):
11 """ Decorator helper class for marking functions as hook specifications.
13 You can instantiate it with a project_name to get a decorator.
14 Calling PluginManager.add_hookspecs later will discover all marked functions
15 if the PluginManager uses the same project_name.
16 """
18 def __init__(self, project_name):
19 self.project_name = project_name
21 def __call__(
22 self, function=None, firstresult=False, historic=False, warn_on_impl=None
23 ):
24 """ if passed a function, directly sets attributes on the function
25 which will make it discoverable to add_hookspecs(). If passed no
26 function, returns a decorator which can be applied to a function
27 later using the attributes supplied.
29 If firstresult is True the 1:N hook call (N being the number of registered
30 hook implementation functions) will stop at I<=N when the I'th function
31 returns a non-None result.
33 If historic is True calls to a hook will be memorized and replayed
34 on later registered plugins.
36 """
38 def setattr_hookspec_opts(func):
39 if historic and firstresult:
40 raise ValueError("cannot have a historic firstresult hook")
41 setattr(
42 func,
43 self.project_name + "_spec",
44 dict(
45 firstresult=firstresult,
46 historic=historic,
47 warn_on_impl=warn_on_impl,
48 ),
49 )
50 return func
52 if function is not None:
53 return setattr_hookspec_opts(function)
54 else:
55 return setattr_hookspec_opts
58class HookimplMarker(object):
59 """ Decorator helper class for marking functions as hook implementations.
61 You can instantiate with a project_name to get a decorator.
62 Calling PluginManager.register later will discover all marked functions
63 if the PluginManager uses the same project_name.
64 """
66 def __init__(self, project_name):
67 self.project_name = project_name
69 def __call__(
70 self,
71 function=None,
72 hookwrapper=False,
73 optionalhook=False,
74 tryfirst=False,
75 trylast=False,
76 ):
78 """ if passed a function, directly sets attributes on the function
79 which will make it discoverable to register(). If passed no function,
80 returns a decorator which can be applied to a function later using
81 the attributes supplied.
83 If optionalhook is True a missing matching hook specification will not result
84 in an error (by default it is an error if no matching spec is found).
86 If tryfirst is True this hook implementation will run as early as possible
87 in the chain of N hook implementations for a specfication.
89 If trylast is True this hook implementation will run as late as possible
90 in the chain of N hook implementations.
92 If hookwrapper is True the hook implementations needs to execute exactly
93 one "yield". The code before the yield is run early before any non-hookwrapper
94 function is run. The code after the yield is run after all non-hookwrapper
95 function have run. The yield receives a ``_Result`` object representing
96 the exception or result outcome of the inner calls (including other hookwrapper
97 calls).
99 """
101 def setattr_hookimpl_opts(func):
102 setattr(
103 func,
104 self.project_name + "_impl",
105 dict(
106 hookwrapper=hookwrapper,
107 optionalhook=optionalhook,
108 tryfirst=tryfirst,
109 trylast=trylast,
110 ),
111 )
112 return func
114 if function is None:
115 return setattr_hookimpl_opts
116 else:
117 return setattr_hookimpl_opts(function)
120def normalize_hookimpl_opts(opts):
121 opts.setdefault("tryfirst", False)
122 opts.setdefault("trylast", False)
123 opts.setdefault("hookwrapper", False)
124 opts.setdefault("optionalhook", False)
127if hasattr(inspect, "getfullargspec"):
129 def _getargspec(func):
130 return inspect.getfullargspec(func)
133else:
135 def _getargspec(func):
136 return inspect.getargspec(func)
139_PYPY3 = hasattr(sys, "pypy_version_info") and sys.version_info.major == 3
142def varnames(func):
143 """Return tuple of positional and keywrord argument names for a function,
144 method, class or callable.
146 In case of a class, its ``__init__`` method is considered.
147 For methods the ``self`` parameter is not included.
148 """
149 cache = getattr(func, "__dict__", {})
150 try:
151 return cache["_varnames"]
152 except KeyError:
153 pass
155 if inspect.isclass(func):
156 try:
157 func = func.__init__
158 except AttributeError:
159 return (), ()
160 elif not inspect.isroutine(func): # callable object?
161 try:
162 func = getattr(func, "__call__", func)
163 except Exception:
164 return ()
166 try: # func MUST be a function or method here or we won't parse any args
167 spec = _getargspec(func)
168 except TypeError:
169 return (), ()
171 args, defaults = tuple(spec.args), spec.defaults
172 if defaults:
173 index = -len(defaults)
174 args, defaults = args[:index], tuple(args[index:])
175 else:
176 defaults = ()
178 # strip any implicit instance arg
179 # pypy3 uses "obj" instead of "self" for default dunder methods
180 implicit_names = ("self",) if not _PYPY3 else ("self", "obj")
181 if args:
182 if inspect.ismethod(func) or (
183 "." in getattr(func, "__qualname__", ()) and args[0] in implicit_names
184 ):
185 args = args[1:]
187 try:
188 cache["_varnames"] = args, defaults
189 except TypeError:
190 pass
191 return args, defaults
194class _HookRelay(object):
195 """ hook holder object for performing 1:N hook calls where N is the number
196 of registered plugins.
198 """
200 def __init__(self, trace):
201 self._trace = trace
204class _HookCaller(object):
205 def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
206 self.name = name
207 self._wrappers = []
208 self._nonwrappers = []
209 self._hookexec = hook_execute
210 self.argnames = None
211 self.kwargnames = None
212 self.multicall = _multicall
213 self.spec = None
214 if specmodule_or_class is not None:
215 assert spec_opts is not None
216 self.set_specification(specmodule_or_class, spec_opts)
218 def has_spec(self):
219 return self.spec is not None
221 def set_specification(self, specmodule_or_class, spec_opts):
222 assert not self.has_spec()
223 self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
224 if spec_opts.get("historic"):
225 self._call_history = []
227 def is_historic(self):
228 return hasattr(self, "_call_history")
230 def _remove_plugin(self, plugin):
231 def remove(wrappers):
232 for i, method in enumerate(wrappers):
233 if method.plugin == plugin:
234 del wrappers[i]
235 return True
237 if remove(self._wrappers) is None:
238 if remove(self._nonwrappers) is None:
239 raise ValueError("plugin %r not found" % (plugin,))
241 def get_hookimpls(self):
242 # Order is important for _hookexec
243 return self._nonwrappers + self._wrappers
245 def _add_hookimpl(self, hookimpl):
246 """Add an implementation to the callback chain.
247 """
248 if hookimpl.hookwrapper:
249 methods = self._wrappers
250 else:
251 methods = self._nonwrappers
253 if hookimpl.trylast:
254 methods.insert(0, hookimpl)
255 elif hookimpl.tryfirst:
256 methods.append(hookimpl)
257 else:
258 # find last non-tryfirst method
259 i = len(methods) - 1
260 while i >= 0 and methods[i].tryfirst:
261 i -= 1
262 methods.insert(i + 1, hookimpl)
264 if "__multicall__" in hookimpl.argnames:
265 warnings.warn(
266 "Support for __multicall__ is now deprecated and will be"
267 "removed in an upcoming release.",
268 DeprecationWarning,
269 )
270 self.multicall = _legacymulticall
272 def __repr__(self):
273 return "<_HookCaller %r>" % (self.name,)
275 def __call__(self, *args, **kwargs):
276 if args:
277 raise TypeError("hook calling supports only keyword arguments")
278 assert not self.is_historic()
279 if self.spec and self.spec.argnames:
280 notincall = (
281 set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
282 )
283 if notincall:
284 warnings.warn(
285 "Argument(s) {} which are declared in the hookspec "
286 "can not be found in this hook call".format(tuple(notincall)),
287 stacklevel=2,
288 )
289 return self._hookexec(self, self.get_hookimpls(), kwargs)
291 def call_historic(self, result_callback=None, kwargs=None, proc=None):
292 """Call the hook with given ``kwargs`` for all registered plugins and
293 for all plugins which will be registered afterwards.
295 If ``result_callback`` is not ``None`` it will be called for for each
296 non-None result obtained from a hook implementation.
298 .. note::
299 The ``proc`` argument is now deprecated.
300 """
301 if proc is not None:
302 warnings.warn(
303 "Support for `proc` argument is now deprecated and will be"
304 "removed in an upcoming release.",
305 DeprecationWarning,
306 )
307 result_callback = proc
309 self._call_history.append((kwargs or {}, result_callback))
310 # historizing hooks don't return results
311 res = self._hookexec(self, self.get_hookimpls(), kwargs)
312 if result_callback is None:
313 return
314 # XXX: remember firstresult isn't compat with historic
315 for x in res or []:
316 result_callback(x)
318 def call_extra(self, methods, kwargs):
319 """ Call the hook with some additional temporarily participating
320 methods using the specified kwargs as call parameters. """
321 old = list(self._nonwrappers), list(self._wrappers)
322 for method in methods:
323 opts = dict(hookwrapper=False, trylast=False, tryfirst=False)
324 hookimpl = HookImpl(None, "<temp>", method, opts)
325 self._add_hookimpl(hookimpl)
326 try:
327 return self(**kwargs)
328 finally:
329 self._nonwrappers, self._wrappers = old
331 def _maybe_apply_history(self, method):
332 """Apply call history to a new hookimpl if it is marked as historic.
333 """
334 if self.is_historic():
335 for kwargs, result_callback in self._call_history:
336 res = self._hookexec(self, [method], kwargs)
337 if res and result_callback is not None:
338 result_callback(res[0])
341class HookImpl(object):
342 def __init__(self, plugin, plugin_name, function, hook_impl_opts):
343 self.function = function
344 self.argnames, self.kwargnames = varnames(self.function)
345 self.plugin = plugin
346 self.opts = hook_impl_opts
347 self.plugin_name = plugin_name
348 self.__dict__.update(hook_impl_opts)
350 def __repr__(self):
351 return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin)
354class HookSpec(object):
355 def __init__(self, namespace, name, opts):
356 self.namespace = namespace
357 self.function = function = getattr(namespace, name)
358 self.name = name
359 self.argnames, self.kwargnames = varnames(function)
360 self.opts = opts
361 self.argnames = ["__multicall__"] + list(self.argnames)
362 self.warn_on_impl = opts.get("warn_on_impl")