Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pluggy/_hooks.py: 70%
Shortcuts on this page
r m x 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"""
5from __future__ import annotations
7import inspect
8import sys
9from types import ModuleType
10from typing import AbstractSet
11from typing import Any
12from typing import Callable
13from typing import Final
14from typing import final
15from typing import Generator
16from typing import List
17from typing import Mapping
18from typing import Optional
19from typing import overload
20from typing import Sequence
21from typing import Tuple
22from typing import TYPE_CHECKING
23from typing import TypedDict
24from typing import TypeVar
25from typing import Union
26import warnings
28from ._result import Result
31_T = TypeVar("_T")
32_F = TypeVar("_F", bound=Callable[..., object])
33_Namespace = Union[ModuleType, type]
34_Plugin = object
35_HookExec = Callable[
36 [str, Sequence["HookImpl"], Mapping[str, object], bool],
37 Union[object, List[object]],
38]
39_HookImplFunction = Callable[..., Union[_T, Generator[None, Result[_T], None]]]
42class HookspecOpts(TypedDict):
43 """Options for a hook specification."""
45 #: Whether the hook is :ref:`first result only <firstresult>`.
46 firstresult: bool
47 #: Whether the hook is :ref:`historic <historic>`.
48 historic: bool
49 #: Whether the hook :ref:`warns when implemented <warn_on_impl>`.
50 warn_on_impl: Warning | None
51 #: Whether the hook warns when :ref:`certain arguments are requested
52 #: <warn_on_impl>`.
53 #:
54 #: .. versionadded:: 1.5
55 warn_on_impl_args: Mapping[str, Warning] | None
58class HookimplOpts(TypedDict):
59 """Options for a hook implementation."""
61 #: Whether the hook implementation is a :ref:`wrapper <hookwrapper>`.
62 wrapper: bool
63 #: Whether the hook implementation is an :ref:`old-style wrapper
64 #: <old_style_hookwrappers>`.
65 hookwrapper: bool
66 #: Whether validation against a hook specification is :ref:`optional
67 #: <optionalhook>`.
68 optionalhook: bool
69 #: Whether to try to order this hook implementation :ref:`first
70 #: <callorder>`.
71 tryfirst: bool
72 #: Whether to try to order this hook implementation :ref:`last
73 #: <callorder>`.
74 trylast: bool
75 #: The name of the hook specification to match, see :ref:`specname`.
76 specname: str | None
79@final
80class HookspecMarker:
81 """Decorator for marking functions as hook specifications.
83 Instantiate it with a project_name to get a decorator.
84 Calling :meth:`PluginManager.add_hookspecs` later will discover all marked
85 functions if the :class:`PluginManager` uses the same project name.
86 """
88 __slots__ = ("project_name",)
90 def __init__(self, project_name: str) -> None:
91 self.project_name: Final = project_name
93 @overload
94 def __call__(
95 self,
96 function: _F,
97 firstresult: bool = False,
98 historic: bool = False,
99 warn_on_impl: Warning | None = None,
100 warn_on_impl_args: Mapping[str, Warning] | None = None,
101 ) -> _F: ...
103 @overload # noqa: F811
104 def __call__( # noqa: F811
105 self,
106 function: None = ...,
107 firstresult: bool = ...,
108 historic: bool = ...,
109 warn_on_impl: Warning | None = ...,
110 warn_on_impl_args: Mapping[str, Warning] | None = ...,
111 ) -> Callable[[_F], _F]: ...
113 def __call__( # noqa: F811
114 self,
115 function: _F | None = None,
116 firstresult: bool = False,
117 historic: bool = False,
118 warn_on_impl: Warning | None = None,
119 warn_on_impl_args: Mapping[str, Warning] | None = None,
120 ) -> _F | Callable[[_F], _F]:
121 """If passed a function, directly sets attributes on the function
122 which will make it discoverable to :meth:`PluginManager.add_hookspecs`.
124 If passed no function, returns a decorator which can be applied to a
125 function later using the attributes supplied.
127 :param firstresult:
128 If ``True``, the 1:N hook call (N being the number of registered
129 hook implementation functions) will stop at I<=N when the I'th
130 function returns a non-``None`` result. See :ref:`firstresult`.
132 :param historic:
133 If ``True``, every call to the hook will be memorized and replayed
134 on plugins registered after the call was made. See :ref:`historic`.
136 :param warn_on_impl:
137 If given, every implementation of this hook will trigger the given
138 warning. See :ref:`warn_on_impl`.
140 :param warn_on_impl_args:
141 If given, every implementation of this hook which requests one of
142 the arguments in the dict will trigger the corresponding warning.
143 See :ref:`warn_on_impl`.
145 .. versionadded:: 1.5
146 """
148 def setattr_hookspec_opts(func: _F) -> _F:
149 if historic and firstresult:
150 raise ValueError("cannot have a historic firstresult hook")
151 opts: HookspecOpts = {
152 "firstresult": firstresult,
153 "historic": historic,
154 "warn_on_impl": warn_on_impl,
155 "warn_on_impl_args": warn_on_impl_args,
156 }
157 setattr(func, self.project_name + "_spec", opts)
158 return func
160 if function is not None:
161 return setattr_hookspec_opts(function)
162 else:
163 return setattr_hookspec_opts
166@final
167class HookimplMarker:
168 """Decorator for marking functions as hook implementations.
170 Instantiate it with a ``project_name`` to get a decorator.
171 Calling :meth:`PluginManager.register` later will discover all marked
172 functions if the :class:`PluginManager` uses the same project name.
173 """
175 __slots__ = ("project_name",)
177 def __init__(self, project_name: str) -> None:
178 self.project_name: Final = project_name
180 @overload
181 def __call__(
182 self,
183 function: _F,
184 hookwrapper: bool = ...,
185 optionalhook: bool = ...,
186 tryfirst: bool = ...,
187 trylast: bool = ...,
188 specname: str | None = ...,
189 wrapper: bool = ...,
190 ) -> _F: ...
192 @overload # noqa: F811
193 def __call__( # noqa: F811
194 self,
195 function: None = ...,
196 hookwrapper: bool = ...,
197 optionalhook: bool = ...,
198 tryfirst: bool = ...,
199 trylast: bool = ...,
200 specname: str | None = ...,
201 wrapper: bool = ...,
202 ) -> Callable[[_F], _F]: ...
204 def __call__( # noqa: F811
205 self,
206 function: _F | None = None,
207 hookwrapper: bool = False,
208 optionalhook: bool = False,
209 tryfirst: bool = False,
210 trylast: bool = False,
211 specname: str | None = None,
212 wrapper: bool = False,
213 ) -> _F | Callable[[_F], _F]:
214 """If passed a function, directly sets attributes on the function
215 which will make it discoverable to :meth:`PluginManager.register`.
217 If passed no function, returns a decorator which can be applied to a
218 function later using the attributes supplied.
220 :param optionalhook:
221 If ``True``, a missing matching hook specification will not result
222 in an error (by default it is an error if no matching spec is
223 found). See :ref:`optionalhook`.
225 :param tryfirst:
226 If ``True``, this hook implementation will run as early as possible
227 in the chain of N hook implementations for a specification. See
228 :ref:`callorder`.
230 :param trylast:
231 If ``True``, this hook implementation will run as late as possible
232 in the chain of N hook implementations for a specification. See
233 :ref:`callorder`.
235 :param wrapper:
236 If ``True`` ("new-style hook wrapper"), the hook implementation
237 needs to execute exactly one ``yield``. The code before the
238 ``yield`` is run early before any non-hook-wrapper function is run.
239 The code after the ``yield`` is run after all non-hook-wrapper
240 functions have run. The ``yield`` receives the result value of the
241 inner calls, or raises the exception of inner calls (including
242 earlier hook wrapper calls). The return value of the function
243 becomes the return value of the hook, and a raised exception becomes
244 the exception of the hook. See :ref:`hookwrapper`.
246 :param hookwrapper:
247 If ``True`` ("old-style hook wrapper"), the hook implementation
248 needs to execute exactly one ``yield``. The code before the
249 ``yield`` is run early before any non-hook-wrapper function is run.
250 The code after the ``yield`` is run after all non-hook-wrapper
251 function have run The ``yield`` receives a :class:`Result` object
252 representing the exception or result outcome of the inner calls
253 (including earlier hook wrapper calls). This option is mutually
254 exclusive with ``wrapper``. See :ref:`old_style_hookwrapper`.
256 :param specname:
257 If provided, the given name will be used instead of the function
258 name when matching this hook implementation to a hook specification
259 during registration. See :ref:`specname`.
261 .. versionadded:: 1.2.0
262 The ``wrapper`` parameter.
263 """
265 def setattr_hookimpl_opts(func: _F) -> _F:
266 opts: HookimplOpts = {
267 "wrapper": wrapper,
268 "hookwrapper": hookwrapper,
269 "optionalhook": optionalhook,
270 "tryfirst": tryfirst,
271 "trylast": trylast,
272 "specname": specname,
273 }
274 setattr(func, self.project_name + "_impl", opts)
275 return func
277 if function is None:
278 return setattr_hookimpl_opts
279 else:
280 return setattr_hookimpl_opts(function)
283def normalize_hookimpl_opts(opts: HookimplOpts) -> None:
284 opts.setdefault("tryfirst", False)
285 opts.setdefault("trylast", False)
286 opts.setdefault("wrapper", False)
287 opts.setdefault("hookwrapper", False)
288 opts.setdefault("optionalhook", False)
289 opts.setdefault("specname", None)
292_PYPY = hasattr(sys, "pypy_version_info")
295def varnames(func: object) -> tuple[tuple[str, ...], tuple[str, ...]]:
296 """Return tuple of positional and keywrord argument names for a function,
297 method, class or callable.
299 In case of a class, its ``__init__`` method is considered.
300 For methods the ``self`` parameter is not included.
301 """
302 if inspect.isclass(func):
303 try:
304 func = func.__init__
305 except AttributeError:
306 return (), ()
307 elif not inspect.isroutine(func): # callable object?
308 try:
309 func = getattr(func, "__call__", func)
310 except Exception:
311 return (), ()
313 try:
314 # func MUST be a function or method here or we won't parse any args.
315 sig = inspect.signature(
316 func.__func__ if inspect.ismethod(func) else func # type:ignore[arg-type]
317 )
318 except TypeError:
319 return (), ()
321 _valid_param_kinds = (
322 inspect.Parameter.POSITIONAL_ONLY,
323 inspect.Parameter.POSITIONAL_OR_KEYWORD,
324 )
325 _valid_params = {
326 name: param
327 for name, param in sig.parameters.items()
328 if param.kind in _valid_param_kinds
329 }
330 args = tuple(_valid_params)
331 defaults = (
332 tuple(
333 param.default
334 for param in _valid_params.values()
335 if param.default is not param.empty
336 )
337 or None
338 )
340 if defaults:
341 index = -len(defaults)
342 args, kwargs = args[:index], tuple(args[index:])
343 else:
344 kwargs = ()
346 # strip any implicit instance arg
347 # pypy3 uses "obj" instead of "self" for default dunder methods
348 if not _PYPY:
349 implicit_names: tuple[str, ...] = ("self",)
350 else:
351 implicit_names = ("self", "obj")
352 if args:
353 qualname: str = getattr(func, "__qualname__", "")
354 if inspect.ismethod(func) or ("." in qualname and args[0] in implicit_names):
355 args = args[1:]
357 return args, kwargs
360@final
361class HookRelay:
362 """Hook holder object for performing 1:N hook calls where N is the number
363 of registered plugins."""
365 __slots__ = ("__dict__",)
367 def __init__(self) -> None:
368 """:meta private:"""
370 if TYPE_CHECKING:
372 def __getattr__(self, name: str) -> HookCaller: ...
375# Historical name (pluggy<=1.2), kept for backward compatibility.
376_HookRelay = HookRelay
379_CallHistory = List[Tuple[Mapping[str, object], Optional[Callable[[Any], None]]]]
382class HookCaller:
383 """A caller of all registered implementations of a hook specification."""
385 __slots__ = (
386 "name",
387 "spec",
388 "_hookexec",
389 "_hookimpls",
390 "_call_history",
391 )
393 def __init__(
394 self,
395 name: str,
396 hook_execute: _HookExec,
397 specmodule_or_class: _Namespace | None = None,
398 spec_opts: HookspecOpts | None = None,
399 ) -> None:
400 """:meta private:"""
401 #: Name of the hook getting called.
402 self.name: Final = name
403 self._hookexec: Final = hook_execute
404 # The hookimpls list. The caller iterates it *in reverse*. Format:
405 # 1. trylast nonwrappers
406 # 2. nonwrappers
407 # 3. tryfirst nonwrappers
408 # 4. trylast wrappers
409 # 5. wrappers
410 # 6. tryfirst wrappers
411 self._hookimpls: Final[list[HookImpl]] = []
412 self._call_history: _CallHistory | None = None
413 # TODO: Document, or make private.
414 self.spec: HookSpec | None = None
415 if specmodule_or_class is not None:
416 assert spec_opts is not None
417 self.set_specification(specmodule_or_class, spec_opts)
419 # TODO: Document, or make private.
420 def has_spec(self) -> bool:
421 return self.spec is not None
423 # TODO: Document, or make private.
424 def set_specification(
425 self,
426 specmodule_or_class: _Namespace,
427 spec_opts: HookspecOpts,
428 ) -> None:
429 if self.spec is not None:
430 raise ValueError(
431 f"Hook {self.spec.name!r} is already registered "
432 f"within namespace {self.spec.namespace}"
433 )
434 self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
435 if spec_opts.get("historic"):
436 self._call_history = []
438 def is_historic(self) -> bool:
439 """Whether this caller is :ref:`historic <historic>`."""
440 return self._call_history is not None
442 def _remove_plugin(self, plugin: _Plugin) -> None:
443 for i, method in enumerate(self._hookimpls):
444 if method.plugin == plugin:
445 del self._hookimpls[i]
446 return
447 raise ValueError(f"plugin {plugin!r} not found")
449 def get_hookimpls(self) -> list[HookImpl]:
450 """Get all registered hook implementations for this hook."""
451 return self._hookimpls.copy()
453 def _add_hookimpl(self, hookimpl: HookImpl) -> None:
454 """Add an implementation to the callback chain."""
455 for i, method in enumerate(self._hookimpls):
456 if method.hookwrapper or method.wrapper:
457 splitpoint = i
458 break
459 else:
460 splitpoint = len(self._hookimpls)
461 if hookimpl.hookwrapper or hookimpl.wrapper:
462 start, end = splitpoint, len(self._hookimpls)
463 else:
464 start, end = 0, splitpoint
466 if hookimpl.trylast:
467 self._hookimpls.insert(start, hookimpl)
468 elif hookimpl.tryfirst:
469 self._hookimpls.insert(end, hookimpl)
470 else:
471 # find last non-tryfirst method
472 i = end - 1
473 while i >= start and self._hookimpls[i].tryfirst:
474 i -= 1
475 self._hookimpls.insert(i + 1, hookimpl)
477 def __repr__(self) -> str:
478 return f"<HookCaller {self.name!r}>"
480 def _verify_all_args_are_provided(self, kwargs: Mapping[str, object]) -> None:
481 # This is written to avoid expensive operations when not needed.
482 if self.spec:
483 for argname in self.spec.argnames:
484 if argname not in kwargs:
485 notincall = ", ".join(
486 repr(argname)
487 for argname in self.spec.argnames
488 # Avoid self.spec.argnames - kwargs.keys() - doesn't preserve order.
489 if argname not in kwargs.keys()
490 )
491 warnings.warn(
492 "Argument(s) {} which are declared in the hookspec "
493 "cannot be found in this hook call".format(notincall),
494 stacklevel=2,
495 )
496 break
498 def __call__(self, **kwargs: object) -> Any:
499 """Call the hook.
501 Only accepts keyword arguments, which should match the hook
502 specification.
504 Returns the result(s) of calling all registered plugins, see
505 :ref:`calling`.
506 """
507 assert (
508 not self.is_historic()
509 ), "Cannot directly call a historic hook - use call_historic instead."
510 self._verify_all_args_are_provided(kwargs)
511 firstresult = self.spec.opts.get("firstresult", False) if self.spec else False
512 # Copy because plugins may register other plugins during iteration (#438).
513 return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
515 def call_historic(
516 self,
517 result_callback: Callable[[Any], None] | None = None,
518 kwargs: Mapping[str, object] | None = None,
519 ) -> None:
520 """Call the hook with given ``kwargs`` for all registered plugins and
521 for all plugins which will be registered afterwards, see
522 :ref:`historic`.
524 :param result_callback:
525 If provided, will be called for each non-``None`` result obtained
526 from a hook implementation.
527 """
528 assert self._call_history is not None
529 kwargs = kwargs or {}
530 self._verify_all_args_are_provided(kwargs)
531 self._call_history.append((kwargs, result_callback))
532 # Historizing hooks don't return results.
533 # Remember firstresult isn't compatible with historic.
534 # Copy because plugins may register other plugins during iteration (#438).
535 res = self._hookexec(self.name, self._hookimpls.copy(), kwargs, False)
536 if result_callback is None:
537 return
538 if isinstance(res, list):
539 for x in res:
540 result_callback(x)
542 def call_extra(
543 self, methods: Sequence[Callable[..., object]], kwargs: Mapping[str, object]
544 ) -> Any:
545 """Call the hook with some additional temporarily participating
546 methods using the specified ``kwargs`` as call parameters, see
547 :ref:`call_extra`."""
548 assert (
549 not self.is_historic()
550 ), "Cannot directly call a historic hook - use call_historic instead."
551 self._verify_all_args_are_provided(kwargs)
552 opts: HookimplOpts = {
553 "wrapper": False,
554 "hookwrapper": False,
555 "optionalhook": False,
556 "trylast": False,
557 "tryfirst": False,
558 "specname": None,
559 }
560 hookimpls = self._hookimpls.copy()
561 for method in methods:
562 hookimpl = HookImpl(None, "<temp>", method, opts)
563 # Find last non-tryfirst nonwrapper method.
564 i = len(hookimpls) - 1
565 while i >= 0 and (
566 # Skip wrappers.
567 (hookimpls[i].hookwrapper or hookimpls[i].wrapper)
568 # Skip tryfirst nonwrappers.
569 or hookimpls[i].tryfirst
570 ):
571 i -= 1
572 hookimpls.insert(i + 1, hookimpl)
573 firstresult = self.spec.opts.get("firstresult", False) if self.spec else False
574 return self._hookexec(self.name, hookimpls, kwargs, firstresult)
576 def _maybe_apply_history(self, method: HookImpl) -> None:
577 """Apply call history to a new hookimpl if it is marked as historic."""
578 if self.is_historic():
579 assert self._call_history is not None
580 for kwargs, result_callback in self._call_history:
581 res = self._hookexec(self.name, [method], kwargs, False)
582 if res and result_callback is not None:
583 # XXX: remember firstresult isn't compat with historic
584 assert isinstance(res, list)
585 result_callback(res[0])
588# Historical name (pluggy<=1.2), kept for backward compatibility.
589_HookCaller = HookCaller
592class _SubsetHookCaller(HookCaller):
593 """A proxy to another HookCaller which manages calls to all registered
594 plugins except the ones from remove_plugins."""
596 # This class is unusual: in inhertits from `HookCaller` so all of
597 # the *code* runs in the class, but it delegates all underlying *data*
598 # to the original HookCaller.
599 # `subset_hook_caller` used to be implemented by creating a full-fledged
600 # HookCaller, copying all hookimpls from the original. This had problems
601 # with memory leaks (#346) and historic calls (#347), which make a proxy
602 # approach better.
603 # An alternative implementation is to use a `_getattr__`/`__getattribute__`
604 # proxy, however that adds more overhead and is more tricky to implement.
606 __slots__ = (
607 "_orig",
608 "_remove_plugins",
609 )
611 def __init__(self, orig: HookCaller, remove_plugins: AbstractSet[_Plugin]) -> None:
612 self._orig = orig
613 self._remove_plugins = remove_plugins
614 self.name = orig.name # type: ignore[misc]
615 self._hookexec = orig._hookexec # type: ignore[misc]
617 @property # type: ignore[misc]
618 def _hookimpls(self) -> list[HookImpl]:
619 return [
620 impl
621 for impl in self._orig._hookimpls
622 if impl.plugin not in self._remove_plugins
623 ]
625 @property
626 def spec(self) -> HookSpec | None: # type: ignore[override]
627 return self._orig.spec
629 @property
630 def _call_history(self) -> _CallHistory | None: # type: ignore[override]
631 return self._orig._call_history
633 def __repr__(self) -> str:
634 return f"<_SubsetHookCaller {self.name!r}>"
637@final
638class HookImpl:
639 """A hook implementation in a :class:`HookCaller`."""
641 __slots__ = (
642 "function",
643 "argnames",
644 "kwargnames",
645 "plugin",
646 "opts",
647 "plugin_name",
648 "wrapper",
649 "hookwrapper",
650 "optionalhook",
651 "tryfirst",
652 "trylast",
653 )
655 def __init__(
656 self,
657 plugin: _Plugin,
658 plugin_name: str,
659 function: _HookImplFunction[object],
660 hook_impl_opts: HookimplOpts,
661 ) -> None:
662 """:meta private:"""
663 #: The hook implementation function.
664 self.function: Final = function
665 argnames, kwargnames = varnames(self.function)
666 #: The positional parameter names of ``function```.
667 self.argnames: Final = argnames
668 #: The keyword parameter names of ``function```.
669 self.kwargnames: Final = kwargnames
670 #: The plugin which defined this hook implementation.
671 self.plugin: Final = plugin
672 #: The :class:`HookimplOpts` used to configure this hook implementation.
673 self.opts: Final = hook_impl_opts
674 #: The name of the plugin which defined this hook implementation.
675 self.plugin_name: Final = plugin_name
676 #: Whether the hook implementation is a :ref:`wrapper <hookwrapper>`.
677 self.wrapper: Final = hook_impl_opts["wrapper"]
678 #: Whether the hook implementation is an :ref:`old-style wrapper
679 #: <old_style_hookwrappers>`.
680 self.hookwrapper: Final = hook_impl_opts["hookwrapper"]
681 #: Whether validation against a hook specification is :ref:`optional
682 #: <optionalhook>`.
683 self.optionalhook: Final = hook_impl_opts["optionalhook"]
684 #: Whether to try to order this hook implementation :ref:`first
685 #: <callorder>`.
686 self.tryfirst: Final = hook_impl_opts["tryfirst"]
687 #: Whether to try to order this hook implementation :ref:`last
688 #: <callorder>`.
689 self.trylast: Final = hook_impl_opts["trylast"]
691 def __repr__(self) -> str:
692 return f"<HookImpl plugin_name={self.plugin_name!r}, plugin={self.plugin!r}>"
695@final
696class HookSpec:
697 __slots__ = (
698 "namespace",
699 "function",
700 "name",
701 "argnames",
702 "kwargnames",
703 "opts",
704 "warn_on_impl",
705 "warn_on_impl_args",
706 )
708 def __init__(self, namespace: _Namespace, name: str, opts: HookspecOpts) -> None:
709 self.namespace = namespace
710 self.function: Callable[..., object] = getattr(namespace, name)
711 self.name = name
712 self.argnames, self.kwargnames = varnames(self.function)
713 self.opts = opts
714 self.warn_on_impl = opts.get("warn_on_impl")
715 self.warn_on_impl_args = opts.get("warn_on_impl_args")