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