From 17d4ee957ffe86e5e65f892fef8898bb2a386d6e Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Wed, 22 Jul 2026 12:24:41 +0200 Subject: [PATCH] chore(structure): split hook modules into role-specific files Move hook types, markers, callers, implementations, and multicall out of the monolithic _hooks/_callers modules so later typed-config and CompletionHook work can land without thrashing one huge file. Keep _hooks and _callers as re-export shims for import compatibility. Co-authored-by: Cursor AI Co-authored-by: Cursor Grok 4.5 --- CHANGELOG.rst | 6 +- changelog/703.trivial.rst | 4 + src/pluggy/_caller.py | 322 +++++++++++++++ src/pluggy/_callers.py | 183 +-------- src/pluggy/_config.py | 54 +++ src/pluggy/_decorators.py | 356 +++++++++++++++++ src/pluggy/_execution.py | 174 +++++++++ src/pluggy/_hooks.py | 798 ++------------------------------------ src/pluggy/_impl.py | 80 ++++ 9 files changed, 1049 insertions(+), 928 deletions(-) create mode 100644 changelog/703.trivial.rst create mode 100644 src/pluggy/_caller.py create mode 100644 src/pluggy/_config.py create mode 100644 src/pluggy/_decorators.py create mode 100644 src/pluggy/_execution.py create mode 100644 src/pluggy/_impl.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 683b399f..061d025a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -222,7 +222,7 @@ Features .. code-block:: python - def my_hook_implementation(arg): + def my_hook_impl(arg): print("before") yield print("after") @@ -230,7 +230,7 @@ Features @hookimpl(hookwrapper=True) def my_hook(arg): - return my_hook_implementation(arg) + return my_hook_impl(arg) change it to use ``yield from`` instead: @@ -238,7 +238,7 @@ Features @hookimpl(hookwrapper=True) def my_hook(arg): - yield from my_hook_implementation(arg) + yield from my_hook_impl(arg) - `#309 `_: Add official support for Python 3.9. diff --git a/changelog/703.trivial.rst b/changelog/703.trivial.rst new file mode 100644 index 00000000..14ec81aa --- /dev/null +++ b/changelog/703.trivial.rst @@ -0,0 +1,4 @@ +The internal ``pluggy._hooks`` module was split into role-specific modules +(``_caller``, ``_config``, ``_decorators``, ``_execution`` and ``_impl``). +``pluggy._hooks`` remains as a re-export shim, and the public API is +unchanged. diff --git a/src/pluggy/_caller.py b/src/pluggy/_caller.py new file mode 100644 index 00000000..52b02316 --- /dev/null +++ b/src/pluggy/_caller.py @@ -0,0 +1,322 @@ +""" +Hook callers and relay. +""" + +from __future__ import annotations + +from collections.abc import Callable +from collections.abc import Mapping +from collections.abc import Sequence +from collections.abc import Set as AbstractSet +from typing import Any +from typing import Final +from typing import final +from typing import TYPE_CHECKING +from typing import TypeAlias +import warnings + +from ._config import HookimplOpts +from ._config import HookspecOpts +from ._decorators import _Namespace +from ._decorators import HookSpec +from ._impl import _Plugin +from ._impl import HookImpl + + +_HookExec: TypeAlias = Callable[ + [str, Sequence[HookImpl], Mapping[str, object], bool], + object | list[object], +] + + +@final +class HookRelay: + """Hook holder object for performing 1:N hook calls where N is the number + of registered plugins.""" + + __slots__ = ("__dict__",) + + def __init__(self) -> None: + """:meta private:""" + + if TYPE_CHECKING: + + def __getattr__(self, name: str) -> HookCaller: ... + + +# Historical name (pluggy<=1.2), kept for backward compatibility. +_HookRelay = HookRelay + + +_CallHistory: TypeAlias = list[ + tuple[Mapping[str, object], Callable[[Any], None] | None] +] + + +class HookCaller: + """A caller of all registered implementations of a hook specification.""" + + __slots__ = ( + "_call_history", + "_hookexec", + "_hookimpls", + "name", + "spec", + ) + + def __init__( + self, + name: str, + hook_execute: _HookExec, + specmodule_or_class: _Namespace | None = None, + spec_opts: HookspecOpts | None = None, + ) -> None: + """:meta private:""" + #: Name of the hook getting called. + self.name: Final = name + self._hookexec: Final = hook_execute + # The hookimpls list. The caller iterates it *in reverse*. Format: + # 1. trylast nonwrappers + # 2. nonwrappers + # 3. tryfirst nonwrappers + # 4. trylast wrappers + # 5. wrappers + # 6. tryfirst wrappers + self._hookimpls: Final[list[HookImpl]] = [] + self._call_history: _CallHistory | None = None + # TODO: Document, or make private. + self.spec: HookSpec | None = None + if specmodule_or_class is not None: + assert spec_opts is not None + self.set_specification(specmodule_or_class, spec_opts) + + # TODO: Document, or make private. + def has_spec(self) -> bool: + return self.spec is not None + + # TODO: Document, or make private. + def set_specification( + self, + specmodule_or_class: _Namespace, + spec_opts: HookspecOpts, + ) -> None: + if self.spec is not None: + raise ValueError( + f"Hook {self.spec.name!r} is already registered " + f"within namespace {self.spec.namespace}" + ) + self.spec = HookSpec(specmodule_or_class, self.name, spec_opts) + if spec_opts.get("historic"): + self._call_history = [] + + def is_historic(self) -> bool: + """Whether this caller is :ref:`historic `.""" + return self._call_history is not None + + def _remove_plugin(self, plugin: _Plugin) -> None: + """Remove all hook implementations registered by the given plugin.""" + remaining = [impl for impl in self._hookimpls if impl.plugin != plugin] + if len(remaining) == len(self._hookimpls): + raise ValueError(f"plugin {plugin!r} not found") + self._hookimpls[:] = remaining + + def get_hookimpls(self) -> list[HookImpl]: + """Get all registered hook implementations for this hook.""" + return self._hookimpls.copy() + + def _add_hookimpl(self, hookimpl: HookImpl) -> None: + """Add an implementation to the callback chain.""" + for i, method in enumerate(self._hookimpls): + if method.hookwrapper or method.wrapper: + splitpoint = i + break + else: + splitpoint = len(self._hookimpls) + if hookimpl.hookwrapper or hookimpl.wrapper: + start, end = splitpoint, len(self._hookimpls) + else: + start, end = 0, splitpoint + + if hookimpl.trylast: + self._hookimpls.insert(start, hookimpl) + elif hookimpl.tryfirst: + self._hookimpls.insert(end, hookimpl) + else: + # find last non-tryfirst method + i = end - 1 + while i >= start and self._hookimpls[i].tryfirst: + i -= 1 + self._hookimpls.insert(i + 1, hookimpl) + + def __repr__(self) -> str: + return f"" + + def _apply_defaults(self, kwargs: Mapping[str, object]) -> Mapping[str, object]: + if self.spec is None or not self.spec.kwargdefaults: + return kwargs + return {**self.spec.kwargdefaults, **kwargs} + + def _verify_all_args_are_provided(self, kwargs: Mapping[str, object]) -> None: + # This is written to avoid expensive operations when not needed. + if self.spec: + for argname in self.spec.argnames: + if argname not in kwargs: + notincall = ", ".join( + repr(argname) + for argname in self.spec.argnames + # Avoid self.spec.argnames - kwargs.keys() + # it doesn't preserve order. + if argname not in kwargs + ) + warnings.warn( + f"Argument(s) {notincall} which are declared in the hookspec " + "cannot be found in this hook call", + # 3, not 2: the warning is raised in this helper, which + # is called by __call__/call_historic/call_extra, which + # are called by the code making the hook call. + stacklevel=3, + ) + break + + def __call__(self, **kwargs: object) -> Any: + """Call the hook. + + Only accepts keyword arguments, which should match the hook + specification. + + Returns the result(s) of calling all registered plugins, see + :ref:`calling`. + """ + assert not self.is_historic(), ( + "Cannot directly call a historic hook - use call_historic instead." + ) + call_kwargs = self._apply_defaults(kwargs) + self._verify_all_args_are_provided(call_kwargs) + firstresult = self.spec.opts.get("firstresult", False) if self.spec else False + # Copy because plugins may register other plugins during iteration (#438). + return self._hookexec( + self.name, self._hookimpls.copy(), call_kwargs, firstresult + ) + + def call_historic( + self, + result_callback: Callable[[Any], None] | None = None, + kwargs: Mapping[str, object] | None = None, + ) -> None: + """Call the hook with given ``kwargs`` for all registered plugins and + for all plugins which will be registered afterwards, see + :ref:`historic`. + + :param result_callback: + If provided, will be called for each non-``None`` result obtained + from a hook implementation. + """ + assert self._call_history is not None + kwargs = kwargs or {} + kwargs = self._apply_defaults(kwargs) + self._verify_all_args_are_provided(kwargs) + self._call_history.append((kwargs, result_callback)) + # Historizing hooks don't return results. + # Remember firstresult isn't compatible with historic. + # Copy because plugins may register other plugins during iteration (#438). + res = self._hookexec(self.name, self._hookimpls.copy(), kwargs, False) + if result_callback is None: + return + if isinstance(res, list): + for x in res: + result_callback(x) + + def call_extra( + self, methods: Sequence[Callable[..., object]], kwargs: Mapping[str, object] + ) -> Any: + """Call the hook with some additional temporarily participating + methods using the specified ``kwargs`` as call parameters, see + :ref:`call_extra`.""" + assert not self.is_historic(), ( + "Cannot directly call a historic hook - use call_historic instead." + ) + kwargs = self._apply_defaults(kwargs) + self._verify_all_args_are_provided(kwargs) + opts: HookimplOpts = { + "wrapper": False, + "hookwrapper": False, + "optionalhook": False, + "trylast": False, + "tryfirst": False, + "specname": None, + } + hookimpls = self._hookimpls.copy() + for method in methods: + hookimpl = HookImpl(None, "", method, opts) + # Find last non-tryfirst nonwrapper method. + i = len(hookimpls) - 1 + while i >= 0 and ( + # Skip wrappers. + (hookimpls[i].hookwrapper or hookimpls[i].wrapper) + # Skip tryfirst nonwrappers. + or hookimpls[i].tryfirst + ): + i -= 1 + hookimpls.insert(i + 1, hookimpl) + firstresult = self.spec.opts.get("firstresult", False) if self.spec else False + return self._hookexec(self.name, hookimpls, kwargs, firstresult) + + def _maybe_apply_history(self, method: HookImpl) -> None: + """Apply call history to a new hookimpl if it is marked as historic.""" + if self.is_historic(): + assert self._call_history is not None + for kwargs, result_callback in self._call_history: + res = self._hookexec(self.name, [method], kwargs, False) + if res and result_callback is not None: + # XXX: remember firstresult isn't compat with historic + assert isinstance(res, list) + result_callback(res[0]) + + +# Historical name (pluggy<=1.2), kept for backward compatibility. +_HookCaller = HookCaller + + +class _SubsetHookCaller(HookCaller): + """A proxy to another HookCaller which manages calls to all registered + plugins except the ones from remove_plugins.""" + + # This class is unusual: in inhertits from `HookCaller` so all of + # the *code* runs in the class, but it delegates all underlying *data* + # to the original HookCaller. + # `subset_hook_caller` used to be implemented by creating a full-fledged + # HookCaller, copying all hookimpls from the original. This had problems + # with memory leaks (#346) and historic calls (#347), which make a proxy + # approach better. + # An alternative implementation is to use a `_getattr__`/`__getattribute__` + # proxy, however that adds more overhead and is more tricky to implement. + + __slots__ = ( + "_orig", + "_remove_plugins", + ) + + def __init__(self, orig: HookCaller, remove_plugins: AbstractSet[_Plugin]) -> None: + self._orig = orig + self._remove_plugins = remove_plugins + self.name = orig.name # type: ignore[misc] + self._hookexec = orig._hookexec # type: ignore[misc] + + @property # type: ignore[misc] + def _hookimpls(self) -> list[HookImpl]: + return [ + impl + for impl in self._orig._hookimpls + if impl.plugin not in self._remove_plugins + ] + + @property + def spec(self) -> HookSpec | None: # type: ignore[override] + return self._orig.spec + + @property + def _call_history(self) -> _CallHistory | None: # type: ignore[override] + return self._orig._call_history + + def __repr__(self) -> str: + return f"<_SubsetHookCaller {self.name!r}>" diff --git a/src/pluggy/_callers.py b/src/pluggy/_callers.py index 8b4b1477..1bde1185 100644 --- a/src/pluggy/_callers.py +++ b/src/pluggy/_callers.py @@ -1,174 +1,23 @@ """ -Call loop machinery +Call loop machinery. + +This module re-exports the execution engine for backward compatibility. +Prefer importing from :mod:`pluggy._execution`. """ from __future__ import annotations -from collections.abc import Generator -from collections.abc import Mapping -from collections.abc import Sequence -from typing import cast -from typing import NoReturn -from typing import TYPE_CHECKING -from typing import TypeAlias -import warnings - -from ._hooks import HookImpl -from ._result import HookCallError -from ._result import Result -from ._warnings import PluggyTeardownRaisedWarning - - -# Need to distinguish between old- and new-style hook wrappers. -# Wrapping with a tuple is the fastest type-safe way I found to do it. -Teardown: TypeAlias = Generator[None, object, object] - - -def run_old_style_hookwrapper( - hook_impl: HookImpl, hook_name: str, args: Sequence[object] -) -> Teardown: - """ - backward compatibility wrapper to run a old style hookwrapper as a wrapper - """ - if TYPE_CHECKING: - teardown = cast(Teardown, hook_impl.function(*args)) - else: - teardown = hook_impl.function(*args) - try: - next(teardown) - except StopIteration: - _raise_wrapfail(teardown, "did not yield") - try: - res = yield - result = Result(res, None) - except BaseException as exc: - result = Result(None, exc) - try: - teardown.send(result) - except StopIteration: - pass - except BaseException as e: - _warn_teardown_exception(hook_name, hook_impl, e) - raise - else: - _raise_wrapfail(teardown, "has second yield") - finally: - teardown.close() - return result.get_result() - - -def _raise_wrapfail( - wrap_controller: Generator[None, object, object], - msg: str, -) -> NoReturn: - co = wrap_controller.gi_code # type: ignore[attr-defined] - raise RuntimeError( - f"wrap_controller at {co.co_name!r} {co.co_filename}:{co.co_firstlineno} {msg}" - ) - - -def _warn_teardown_exception( - hook_name: str, hook_impl: HookImpl, e: BaseException -) -> None: - msg = ( - f"A plugin raised an exception during an old-style hookwrapper teardown.\n" - f"Plugin: {hook_impl.plugin_name}, Hook: {hook_name}\n" - f"{type(e).__name__}: {e}\n" - f"For more information see https://pluggy.readthedocs.io/en/stable/api_reference.html#pluggy.PluggyTeardownRaisedWarning" - ) - warnings.warn(PluggyTeardownRaisedWarning(msg), stacklevel=6) - - -def _multicall( - hook_name: str, - hook_impls: Sequence[HookImpl], - caller_kwargs: Mapping[str, object], - firstresult: bool, -) -> object | list[object]: - """Execute a call into multiple python functions/methods and return the - result(s). - - ``caller_kwargs`` comes from HookCaller.__call__(). - """ - __tracebackhide__ = True - results: list[object] = [] - exception = None - teardowns: list[Teardown] = [] - try: # run impl and wrapper setup functions in a loop - for hook_impl in reversed(hook_impls): - try: - args = [caller_kwargs[argname] for argname in hook_impl.argnames] - except KeyError as e: - raise HookCallError( - f"hook call must provide argument {e.args[0]!r}" - ) from e - - if hook_impl.hookwrapper: - function_gen = run_old_style_hookwrapper(hook_impl, hook_name, args) - - next(function_gen) # first yield - teardowns.append(function_gen) - - elif hook_impl.wrapper: - res = hook_impl.function(*args) - # If this cast is not valid, a type error is raised below, - # which is the desired response. - if TYPE_CHECKING: - function_gen = cast(Generator[None, object, object], res) - else: - function_gen = res - try: - next(function_gen) # first yield - except StopIteration: - _raise_wrapfail(function_gen, "did not yield") - teardowns.append(function_gen) - else: - res = hook_impl.function(*args) - if res is not None: - results.append(res) - if firstresult: # halt further impl calls - break - except BaseException as exc: - exception = exc - finally: - if firstresult: # first result hooks return a single value - result = results[0] if results else None - else: - result = results +from ._execution import _multicall +from ._execution import _raise_wrapfail +from ._execution import _warn_teardown_exception +from ._execution import run_old_style_hookwrapper +from ._execution import Teardown - # run all wrapper post-yield blocks - for teardown in reversed(teardowns): - try: - if exception is not None: - try: - teardown.throw(exception) - except RuntimeError as re: - # StopIteration from generator causes RuntimeError - # even for coroutine usage - see #544 - if ( - isinstance(exception, StopIteration) - and re.__cause__ is exception - ): - teardown.close() - continue - else: - raise - else: - teardown.send(result) - # Following is unreachable for a well behaved hook wrapper. - # Try to force finalizers otherwise postponed till GC action. - # Note: close() may raise if generator handles GeneratorExit. - teardown.close() - except StopIteration as si: - result = si.value - exception = None - continue - except BaseException as e: - exception = e - continue - _raise_wrapfail(teardown, "has second yield") - if exception is not None: - raise exception - else: - return result +__all__ = [ + "Teardown", + "_multicall", + "_raise_wrapfail", + "_warn_teardown_exception", + "run_old_style_hookwrapper", +] diff --git a/src/pluggy/_config.py b/src/pluggy/_config.py new file mode 100644 index 00000000..576a7794 --- /dev/null +++ b/src/pluggy/_config.py @@ -0,0 +1,54 @@ +""" +Configuration types for hook specifications and implementations. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import TypedDict + + +class HookspecOpts(TypedDict): + """Options for a hook specification.""" + + #: Whether the hook is :ref:`first result only `. + firstresult: bool + #: Whether the hook is :ref:`historic `. + historic: bool + #: Whether the hook :ref:`warns when implemented `. + warn_on_impl: Warning | None + #: Whether the hook warns when :ref:`certain arguments are requested + #: `. + #: + #: .. versionadded:: 1.5 + warn_on_impl_args: Mapping[str, Warning] | None + + +class HookimplOpts(TypedDict): + """Options for a hook implementation.""" + + #: Whether the hook implementation is a :ref:`wrapper `. + wrapper: bool + #: Whether the hook implementation is an :ref:`old-style wrapper + #: `. + hookwrapper: bool + #: Whether validation against a hook specification is :ref:`optional + #: `. + optionalhook: bool + #: Whether to try to order this hook implementation :ref:`first + #: `. + tryfirst: bool + #: Whether to try to order this hook implementation :ref:`last + #: `. + trylast: bool + #: The name of the hook specification to match, see :ref:`specname`. + specname: str | None + + +def normalize_hookimpl_opts(opts: HookimplOpts) -> None: + opts.setdefault("tryfirst", False) + opts.setdefault("trylast", False) + opts.setdefault("wrapper", False) + opts.setdefault("hookwrapper", False) + opts.setdefault("optionalhook", False) + opts.setdefault("specname", None) diff --git a/src/pluggy/_decorators.py b/src/pluggy/_decorators.py new file mode 100644 index 00000000..c5574a67 --- /dev/null +++ b/src/pluggy/_decorators.py @@ -0,0 +1,356 @@ +""" +Hook markers, specifications, and related helpers. +""" + +from __future__ import annotations + +from collections.abc import Callable +from collections.abc import Mapping +import inspect +import sys +import types +from types import ModuleType +from typing import Final +from typing import final +from typing import overload +from typing import TypeAlias +from typing import TypeVar +import warnings + +from ._config import HookimplOpts +from ._config import HookspecOpts + + +_F = TypeVar("_F", bound=Callable[..., object]) + +_Namespace: TypeAlias = ModuleType | type + + +@final +class HookspecMarker: + """Decorator for marking functions as hook specifications. + + Instantiate it with a project_name to get a decorator. + Calling :meth:`PluginManager.add_hookspecs` later will discover all marked + functions if the :class:`PluginManager` uses the same project name. + """ + + __slots__ = ("project_name",) + + def __init__(self, project_name: str) -> None: + self.project_name: Final = project_name + + @overload + def __call__( + self, + function: _F, + firstresult: bool = False, + historic: bool = False, + warn_on_impl: Warning | None = None, + warn_on_impl_args: Mapping[str, Warning] | None = None, + ) -> _F: ... + + @overload + def __call__( + self, + function: None = ..., + firstresult: bool = ..., + historic: bool = ..., + warn_on_impl: Warning | None = ..., + warn_on_impl_args: Mapping[str, Warning] | None = ..., + ) -> Callable[[_F], _F]: ... + + def __call__( + self, + function: _F | None = None, + firstresult: bool = False, + historic: bool = False, + warn_on_impl: Warning | None = None, + warn_on_impl_args: Mapping[str, Warning] | None = None, + ) -> _F | Callable[[_F], _F]: + """If passed a function, directly sets attributes on the function + which will make it discoverable to :meth:`PluginManager.add_hookspecs`. + + If passed no function, returns a decorator which can be applied to a + function later using the attributes supplied. + + :param firstresult: + If ``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. See :ref:`firstresult`. + + :param historic: + If ``True``, every call to the hook will be memorized and replayed + on plugins registered after the call was made. See :ref:`historic`. + + :param warn_on_impl: + If given, every implementation of this hook will trigger the given + warning. See :ref:`warn_on_impl`. + + :param warn_on_impl_args: + If given, every implementation of this hook which requests one of + the arguments in the dict will trigger the corresponding warning. + See :ref:`warn_on_impl`. + + .. versionadded:: 1.5 + """ + + def setattr_hookspec_opts(func: _F) -> _F: + if historic and firstresult: + raise ValueError("cannot have a historic firstresult hook") + opts: HookspecOpts = { + "firstresult": firstresult, + "historic": historic, + "warn_on_impl": warn_on_impl, + "warn_on_impl_args": warn_on_impl_args, + } + setattr(func, self.project_name + "_spec", opts) + return func + + if function is not None: + return setattr_hookspec_opts(function) + else: + return setattr_hookspec_opts + + +@final +class HookimplMarker: + """Decorator for marking functions as hook implementations. + + Instantiate it with a ``project_name`` to get a decorator. + Calling :meth:`PluginManager.register` later will discover all marked + functions if the :class:`PluginManager` uses the same project name. + """ + + __slots__ = ("project_name",) + + def __init__(self, project_name: str) -> None: + self.project_name: Final = project_name + + @overload + def __call__( + self, + function: _F, + hookwrapper: bool = ..., + optionalhook: bool = ..., + tryfirst: bool = ..., + trylast: bool = ..., + specname: str | None = ..., + wrapper: bool = ..., + ) -> _F: ... + + @overload + def __call__( + self, + function: None = ..., + hookwrapper: bool = ..., + optionalhook: bool = ..., + tryfirst: bool = ..., + trylast: bool = ..., + specname: str | None = ..., + wrapper: bool = ..., + ) -> Callable[[_F], _F]: ... + + def __call__( + self, + function: _F | None = None, + hookwrapper: bool = False, + optionalhook: bool = False, + tryfirst: bool = False, + trylast: bool = False, + specname: str | None = None, + wrapper: bool = False, + ) -> _F | Callable[[_F], _F]: + """If passed a function, directly sets attributes on the function + which will make it discoverable to :meth:`PluginManager.register`. + + If passed no function, returns a decorator which can be applied to a + function later using the attributes supplied. + + :param optionalhook: + If ``True``, a missing matching hook specification will not result + in an error (by default it is an error if no matching spec is + found). See :ref:`optionalhook`. + + :param tryfirst: + If ``True``, this hook implementation will run as early as possible + in the chain of N hook implementations for a specification. See + :ref:`callorder`. + + :param trylast: + If ``True``, this hook implementation will run as late as possible + in the chain of N hook implementations for a specification. See + :ref:`callorder`. + + :param wrapper: + If ``True`` ("new-style hook wrapper"), the hook implementation + needs to execute exactly one ``yield``. The code before the + ``yield`` is run early before any non-hook-wrapper function is run. + The code after the ``yield`` is run after all non-hook-wrapper + functions have run. The ``yield`` receives the result value of the + inner calls, or raises the exception of inner calls (including + earlier hook wrapper calls). The return value of the function + becomes the return value of the hook, and a raised exception becomes + the exception of the hook. See :ref:`hookwrapper`. + + :param hookwrapper: + If ``True`` ("old-style hook wrapper"), the hook implementation + needs to execute exactly one ``yield``. The code before the + ``yield`` is run early before any non-hook-wrapper function is run. + The code after the ``yield`` is run after all non-hook-wrapper + function have run The ``yield`` receives a :class:`Result` object + representing the exception or result outcome of the inner calls + (including earlier hook wrapper calls). This option is mutually + exclusive with ``wrapper``. See :ref:`old_style_hookwrapper`. + + :param specname: + If provided, the given name will be used instead of the function + name when matching this hook implementation to a hook specification + during registration. See :ref:`specname`. + + .. versionadded:: 1.2.0 + The ``wrapper`` parameter. + """ + + def setattr_hookimpl_opts(func: _F) -> _F: + opts: HookimplOpts = { + "wrapper": wrapper, + "hookwrapper": hookwrapper, + "optionalhook": optionalhook, + "tryfirst": tryfirst, + "trylast": trylast, + "specname": specname, + } + setattr(func, self.project_name + "_impl", opts) + return func + + if function is None: + return setattr_hookimpl_opts + else: + return setattr_hookimpl_opts(function) + + +_PYPY = sys.implementation.name == "pypy" +_IMPLICIT_NAMES = ("self", "cls", "obj") if _PYPY else ("self", "cls") + +# Qualnames whose missing-self deprecation warning is suppressed because +# their upstream code is already fixed but not yet released. +# Remove entries once a release with the fix is available. +_NOSELF_WARN_SUPPRESS: frozenset[str] = frozenset( + { + # pytest-timeout >=2.3.2 has the fix, but is unreleased as of 2026-05. + "TimeoutHooks.pytest_timeout_set_timer", + "TimeoutHooks.pytest_timeout_cancel_timer", + } +) + + +def varnames( + func: object, *, legacy_noself: bool = False +) -> tuple[tuple[str, ...], tuple[str, ...]]: + """Return tuple of positional and keyword parameter names for a callable. + + In case of a class, its ``__init__`` method is considered. + For bound methods, the already-bound first parameter is not included. + For unbound methods with a dotted ``__qualname__``, the first parameter is + stripped only if its name is a known implicit name (``self``, ``cls``). + Keyword-only parameters are not included. + + :param legacy_noself: + If ``True``, support hookspec classes whose methods omit ``self``. + When the function looks like a class method but has no implicit first + parameter, a :class:`DeprecationWarning` is emitted. + """ + is_bound = False + if inspect.isclass(func): + try: + func = func.__init__ + except AttributeError: # pragma: no cover - pypy special case + return (), () + is_bound = True + elif not inspect.isroutine(func): # callable object? + try: + # Not a `callable()` check: the `__call__` attribute itself is + # wanted, so that its signature can be inspected below. + func = getattr(func, "__call__", func) # noqa: B004 + except Exception: # pragma: no cover - pypy special case + return (), () + + # Track bound methods before unwrapping, since __func__ loses that info. + if inspect.ismethod(func): + is_bound = True + func = inspect.unwrap(func) # type: ignore[arg-type] + if inspect.ismethod(func): + is_bound = True + func = func.__func__ + + try: + code: types.CodeType = func.__code__ # type: ignore[attr-defined] + defaults: tuple[object, ...] | None = func.__defaults__ # type: ignore[attr-defined] + qualname: str = func.__qualname__ # type: ignore[attr-defined] + except AttributeError: # pragma: no cover + return (), () + + # Get positional argument names (positional-only + positional-or-keyword) + args: tuple[str, ...] = code.co_varnames[: code.co_argcount] + + # Determine which args have defaults + kwargs: tuple[str, ...] + if defaults: + index = -len(defaults) + args, kwargs = args[:index], args[index:] + else: + kwargs = () + + # Strip implicit instance/class arg. + # Check if this looks like a method defined in a class by examining the + # qualname after the last "." segment (if any). A remaining dot + # means it's a class method (e.g. "MyClass.method" or + # "func..MyClass.method"), not just a nested function. + _tail = qualname.rsplit(".", maxsplit=1)[-1] + _is_class_method = "." in _tail + if args: + if is_bound or (_is_class_method and args[0] in _IMPLICIT_NAMES): + args = args[1:] + elif _is_class_method and legacy_noself and _tail not in _NOSELF_WARN_SUPPRESS: + warnings.warn( + f"{qualname} is a method but its first parameter" + f" {args[0]!r} is not 'self'." + f" Add 'self' as the first parameter or use @staticmethod." + f" This will become an error in a future version of pluggy.", + DeprecationWarning, + stacklevel=2, + ) + + return args, kwargs + + +@final +class HookSpec: + __slots__ = ( + "argnames", + "function", + "kwargdefaults", + "kwargnames", + "name", + "namespace", + "opts", + "warn_on_impl", + "warn_on_impl_args", + ) + + def __init__(self, namespace: _Namespace, name: str, opts: HookspecOpts) -> None: + self.namespace = namespace + self.name = name + self.function: Callable[..., object] = getattr(namespace, name) + legacy_noself = inspect.isclass(namespace) and not isinstance( + inspect.getattr_static(namespace, name), staticmethod + ) + self.argnames, self.kwargnames = varnames( + self.function, legacy_noself=legacy_noself + ) + defaults = inspect.unwrap(self.function).__defaults__ + self.kwargdefaults = dict(zip(self.kwargnames, defaults or ())) + self.opts = opts + self.warn_on_impl = opts.get("warn_on_impl") + self.warn_on_impl_args = opts.get("warn_on_impl_args") diff --git a/src/pluggy/_execution.py b/src/pluggy/_execution.py new file mode 100644 index 00000000..cda5210d --- /dev/null +++ b/src/pluggy/_execution.py @@ -0,0 +1,174 @@ +""" +Hook call execution (multicall) machinery. +""" + +from __future__ import annotations + +from collections.abc import Generator +from collections.abc import Mapping +from collections.abc import Sequence +from typing import cast +from typing import NoReturn +from typing import TYPE_CHECKING +from typing import TypeAlias +import warnings + +from ._impl import HookImpl +from ._result import HookCallError +from ._result import Result +from ._warnings import PluggyTeardownRaisedWarning + + +# Need to distinguish between old- and new-style hook wrappers. +# Wrapping with a tuple is the fastest type-safe way I found to do it. +Teardown: TypeAlias = Generator[None, object, object] + + +def run_old_style_hookwrapper( + hook_impl: HookImpl, hook_name: str, args: Sequence[object] +) -> Teardown: + """ + backward compatibility wrapper to run a old style hookwrapper as a wrapper + """ + if TYPE_CHECKING: + teardown = cast(Teardown, hook_impl.function(*args)) + else: + teardown = hook_impl.function(*args) + try: + next(teardown) + except StopIteration: + _raise_wrapfail(teardown, "did not yield") + try: + res = yield + result = Result(res, None) + except BaseException as exc: + result = Result(None, exc) + try: + teardown.send(result) + except StopIteration: + pass + except BaseException as e: + _warn_teardown_exception(hook_name, hook_impl, e) + raise + else: + _raise_wrapfail(teardown, "has second yield") + finally: + teardown.close() + return result.get_result() + + +def _raise_wrapfail( + wrap_controller: Generator[None, object, object], + msg: str, +) -> NoReturn: + co = wrap_controller.gi_code # type: ignore[attr-defined] + raise RuntimeError( + f"wrap_controller at {co.co_name!r} {co.co_filename}:{co.co_firstlineno} {msg}" + ) + + +def _warn_teardown_exception( + hook_name: str, hook_impl: HookImpl, e: BaseException +) -> None: + msg = ( + f"A plugin raised an exception during an old-style hookwrapper teardown.\n" + f"Plugin: {hook_impl.plugin_name}, Hook: {hook_name}\n" + f"{type(e).__name__}: {e}\n" + f"For more information see https://pluggy.readthedocs.io/en/stable/api_reference.html#pluggy.PluggyTeardownRaisedWarning" + ) + warnings.warn(PluggyTeardownRaisedWarning(msg), stacklevel=6) + + +def _multicall( + hook_name: str, + hook_impls: Sequence[HookImpl], + caller_kwargs: Mapping[str, object], + firstresult: bool, +) -> object | list[object]: + """Execute a call into multiple python functions/methods and return the + result(s). + + ``caller_kwargs`` comes from HookCaller.__call__(). + """ + __tracebackhide__ = True + results: list[object] = [] + exception = None + teardowns: list[Teardown] = [] + try: # run impl and wrapper setup functions in a loop + for hook_impl in reversed(hook_impls): + try: + args = [caller_kwargs[argname] for argname in hook_impl.argnames] + except KeyError as e: + raise HookCallError( + f"hook call must provide argument {e.args[0]!r}" + ) from e + + if hook_impl.hookwrapper: + function_gen = run_old_style_hookwrapper(hook_impl, hook_name, args) + + next(function_gen) # first yield + teardowns.append(function_gen) + + elif hook_impl.wrapper: + res = hook_impl.function(*args) + # If this cast is not valid, a type error is raised below, + # which is the desired response. + if TYPE_CHECKING: + function_gen = cast(Generator[None, object, object], res) + else: + function_gen = res + try: + next(function_gen) # first yield + except StopIteration: + _raise_wrapfail(function_gen, "did not yield") + teardowns.append(function_gen) + else: + res = hook_impl.function(*args) + if res is not None: + results.append(res) + if firstresult: # halt further impl calls + break + except BaseException as exc: + exception = exc + finally: + if firstresult: # first result hooks return a single value + result = results[0] if results else None + else: + result = results + + # run all wrapper post-yield blocks + for teardown in reversed(teardowns): + try: + if exception is not None: + try: + teardown.throw(exception) + except RuntimeError as re: + # StopIteration from generator causes RuntimeError + # even for coroutine usage - see #544 + if ( + isinstance(exception, StopIteration) + and re.__cause__ is exception + ): + teardown.close() + continue + else: + raise + else: + teardown.send(result) + # Following is unreachable for a well behaved hook wrapper. + # Try to force finalizers otherwise postponed till GC action. + # Note: close() may raise if generator handles GeneratorExit. + teardown.close() + except StopIteration as si: + result = si.value + exception = None + continue + except BaseException as e: + exception = e + continue + _raise_wrapfail(teardown, "has second yield") + + if exception is not None: + raise exception + else: + return result diff --git a/src/pluggy/_hooks.py b/src/pluggy/_hooks.py index 3c1eaaaa..b57aae66 100644 --- a/src/pluggy/_hooks.py +++ b/src/pluggy/_hooks.py @@ -1,765 +1,47 @@ """ Internal hook annotation, representation and calling machinery. + +This module re-exports symbols from the role-specific modules for +backward compatibility. """ from __future__ import annotations -from collections.abc import Callable -from collections.abc import Generator -from collections.abc import Mapping -from collections.abc import Sequence -from collections.abc import Set as AbstractSet -import inspect -import sys -import types -from types import ModuleType -from typing import Any -from typing import Final -from typing import final -from typing import overload -from typing import TYPE_CHECKING -from typing import TypeAlias -from typing import TypedDict -from typing import TypeVar -import warnings - -from ._result import Result - - -_T = TypeVar("_T") -_F = TypeVar("_F", bound=Callable[..., object]) - -_Namespace: TypeAlias = ModuleType | type -_Plugin: TypeAlias = object -_HookExec: TypeAlias = Callable[ - [str, Sequence["HookImpl"], Mapping[str, object], bool], - object | list[object], +from ._caller import _HookCaller +from ._caller import _HookExec +from ._caller import _HookRelay +from ._caller import _SubsetHookCaller +from ._caller import HookCaller +from ._caller import HookRelay +from ._config import HookimplOpts +from ._config import HookspecOpts +from ._config import normalize_hookimpl_opts +from ._decorators import _Namespace +from ._decorators import HookimplMarker +from ._decorators import HookSpec +from ._decorators import HookspecMarker +from ._decorators import varnames +from ._impl import _HookImplFunction +from ._impl import _Plugin +from ._impl import HookImpl + + +__all__ = [ + "HookCaller", + "HookImpl", + "HookRelay", + "HookSpec", + "HookimplMarker", + "HookimplOpts", + "HookspecMarker", + "HookspecOpts", + "_HookCaller", + "_HookExec", + "_HookImplFunction", + "_HookRelay", + "_Namespace", + "_Plugin", + "_SubsetHookCaller", + "normalize_hookimpl_opts", + "varnames", ] -_HookImplFunction: TypeAlias = Callable[..., _T | Generator[None, Result[_T], None]] - - -class HookspecOpts(TypedDict): - """Options for a hook specification.""" - - #: Whether the hook is :ref:`first result only `. - firstresult: bool - #: Whether the hook is :ref:`historic `. - historic: bool - #: Whether the hook :ref:`warns when implemented `. - warn_on_impl: Warning | None - #: Whether the hook warns when :ref:`certain arguments are requested - #: `. - #: - #: .. versionadded:: 1.5 - warn_on_impl_args: Mapping[str, Warning] | None - - -class HookimplOpts(TypedDict): - """Options for a hook implementation.""" - - #: Whether the hook implementation is a :ref:`wrapper `. - wrapper: bool - #: Whether the hook implementation is an :ref:`old-style wrapper - #: `. - hookwrapper: bool - #: Whether validation against a hook specification is :ref:`optional - #: `. - optionalhook: bool - #: Whether to try to order this hook implementation :ref:`first - #: `. - tryfirst: bool - #: Whether to try to order this hook implementation :ref:`last - #: `. - trylast: bool - #: The name of the hook specification to match, see :ref:`specname`. - specname: str | None - - -@final -class HookspecMarker: - """Decorator for marking functions as hook specifications. - - Instantiate it with a project_name to get a decorator. - Calling :meth:`PluginManager.add_hookspecs` later will discover all marked - functions if the :class:`PluginManager` uses the same project name. - """ - - __slots__ = ("project_name",) - - def __init__(self, project_name: str) -> None: - self.project_name: Final = project_name - - @overload - def __call__( - self, - function: _F, - firstresult: bool = False, - historic: bool = False, - warn_on_impl: Warning | None = None, - warn_on_impl_args: Mapping[str, Warning] | None = None, - ) -> _F: ... - - @overload - def __call__( - self, - function: None = ..., - firstresult: bool = ..., - historic: bool = ..., - warn_on_impl: Warning | None = ..., - warn_on_impl_args: Mapping[str, Warning] | None = ..., - ) -> Callable[[_F], _F]: ... - - def __call__( - self, - function: _F | None = None, - firstresult: bool = False, - historic: bool = False, - warn_on_impl: Warning | None = None, - warn_on_impl_args: Mapping[str, Warning] | None = None, - ) -> _F | Callable[[_F], _F]: - """If passed a function, directly sets attributes on the function - which will make it discoverable to :meth:`PluginManager.add_hookspecs`. - - If passed no function, returns a decorator which can be applied to a - function later using the attributes supplied. - - :param firstresult: - If ``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. See :ref:`firstresult`. - - :param historic: - If ``True``, every call to the hook will be memorized and replayed - on plugins registered after the call was made. See :ref:`historic`. - - :param warn_on_impl: - If given, every implementation of this hook will trigger the given - warning. See :ref:`warn_on_impl`. - - :param warn_on_impl_args: - If given, every implementation of this hook which requests one of - the arguments in the dict will trigger the corresponding warning. - See :ref:`warn_on_impl`. - - .. versionadded:: 1.5 - """ - - def setattr_hookspec_opts(func: _F) -> _F: - if historic and firstresult: - raise ValueError("cannot have a historic firstresult hook") - opts: HookspecOpts = { - "firstresult": firstresult, - "historic": historic, - "warn_on_impl": warn_on_impl, - "warn_on_impl_args": warn_on_impl_args, - } - setattr(func, self.project_name + "_spec", opts) - return func - - if function is not None: - return setattr_hookspec_opts(function) - else: - return setattr_hookspec_opts - - -@final -class HookimplMarker: - """Decorator for marking functions as hook implementations. - - Instantiate it with a ``project_name`` to get a decorator. - Calling :meth:`PluginManager.register` later will discover all marked - functions if the :class:`PluginManager` uses the same project name. - """ - - __slots__ = ("project_name",) - - def __init__(self, project_name: str) -> None: - self.project_name: Final = project_name - - @overload - def __call__( - self, - function: _F, - hookwrapper: bool = ..., - optionalhook: bool = ..., - tryfirst: bool = ..., - trylast: bool = ..., - specname: str | None = ..., - wrapper: bool = ..., - ) -> _F: ... - - @overload - def __call__( - self, - function: None = ..., - hookwrapper: bool = ..., - optionalhook: bool = ..., - tryfirst: bool = ..., - trylast: bool = ..., - specname: str | None = ..., - wrapper: bool = ..., - ) -> Callable[[_F], _F]: ... - - def __call__( - self, - function: _F | None = None, - hookwrapper: bool = False, - optionalhook: bool = False, - tryfirst: bool = False, - trylast: bool = False, - specname: str | None = None, - wrapper: bool = False, - ) -> _F | Callable[[_F], _F]: - """If passed a function, directly sets attributes on the function - which will make it discoverable to :meth:`PluginManager.register`. - - If passed no function, returns a decorator which can be applied to a - function later using the attributes supplied. - - :param optionalhook: - If ``True``, a missing matching hook specification will not result - in an error (by default it is an error if no matching spec is - found). See :ref:`optionalhook`. - - :param tryfirst: - If ``True``, this hook implementation will run as early as possible - in the chain of N hook implementations for a specification. See - :ref:`callorder`. - - :param trylast: - If ``True``, this hook implementation will run as late as possible - in the chain of N hook implementations for a specification. See - :ref:`callorder`. - - :param wrapper: - If ``True`` ("new-style hook wrapper"), the hook implementation - needs to execute exactly one ``yield``. The code before the - ``yield`` is run early before any non-hook-wrapper function is run. - The code after the ``yield`` is run after all non-hook-wrapper - functions have run. The ``yield`` receives the result value of the - inner calls, or raises the exception of inner calls (including - earlier hook wrapper calls). The return value of the function - becomes the return value of the hook, and a raised exception becomes - the exception of the hook. See :ref:`hookwrapper`. - - :param hookwrapper: - If ``True`` ("old-style hook wrapper"), the hook implementation - needs to execute exactly one ``yield``. The code before the - ``yield`` is run early before any non-hook-wrapper function is run. - The code after the ``yield`` is run after all non-hook-wrapper - function have run The ``yield`` receives a :class:`Result` object - representing the exception or result outcome of the inner calls - (including earlier hook wrapper calls). This option is mutually - exclusive with ``wrapper``. See :ref:`old_style_hookwrapper`. - - :param specname: - If provided, the given name will be used instead of the function - name when matching this hook implementation to a hook specification - during registration. See :ref:`specname`. - - .. versionadded:: 1.2.0 - The ``wrapper`` parameter. - """ - - def setattr_hookimpl_opts(func: _F) -> _F: - opts: HookimplOpts = { - "wrapper": wrapper, - "hookwrapper": hookwrapper, - "optionalhook": optionalhook, - "tryfirst": tryfirst, - "trylast": trylast, - "specname": specname, - } - setattr(func, self.project_name + "_impl", opts) - return func - - if function is None: - return setattr_hookimpl_opts - else: - return setattr_hookimpl_opts(function) - - -def normalize_hookimpl_opts(opts: HookimplOpts) -> None: - opts.setdefault("tryfirst", False) - opts.setdefault("trylast", False) - opts.setdefault("wrapper", False) - opts.setdefault("hookwrapper", False) - opts.setdefault("optionalhook", False) - opts.setdefault("specname", None) - - -_PYPY = sys.implementation.name == "pypy" -_IMPLICIT_NAMES = ("self", "cls", "obj") if _PYPY else ("self", "cls") - -# Qualnames whose missing-self deprecation warning is suppressed because -# their upstream code is already fixed but not yet released. -# Remove entries once a release with the fix is available. -_NOSELF_WARN_SUPPRESS: frozenset[str] = frozenset( - { - # pytest-timeout >=2.3.2 has the fix, but is unreleased as of 2026-05. - "TimeoutHooks.pytest_timeout_set_timer", - "TimeoutHooks.pytest_timeout_cancel_timer", - } -) - - -def varnames( - func: object, *, legacy_noself: bool = False -) -> tuple[tuple[str, ...], tuple[str, ...]]: - """Return tuple of positional and keyword parameter names for a callable. - - In case of a class, its ``__init__`` method is considered. - For bound methods, the already-bound first parameter is not included. - For unbound methods with a dotted ``__qualname__``, the first parameter is - stripped only if its name is a known implicit name (``self``, ``cls``). - Keyword-only parameters are not included. - - :param legacy_noself: - If ``True``, support hookspec classes whose methods omit ``self``. - When the function looks like a class method but has no implicit first - parameter, a :class:`DeprecationWarning` is emitted. - """ - is_bound = False - if inspect.isclass(func): - try: - func = func.__init__ - except AttributeError: # pragma: no cover - pypy special case - return (), () - is_bound = True - elif not inspect.isroutine(func): # callable object? - try: - # Not a `callable()` check: the `__call__` attribute itself is - # wanted, so that its signature can be inspected below. - func = getattr(func, "__call__", func) # noqa: B004 - except Exception: # pragma: no cover - pypy special case - return (), () - - # Track bound methods before unwrapping, since __func__ loses that info. - if inspect.ismethod(func): - is_bound = True - func = inspect.unwrap(func) # type: ignore[arg-type] - if inspect.ismethod(func): - is_bound = True - func = func.__func__ - - try: - code: types.CodeType = func.__code__ # type: ignore[attr-defined] - defaults: tuple[object, ...] | None = func.__defaults__ # type: ignore[attr-defined] - qualname: str = func.__qualname__ # type: ignore[attr-defined] - except AttributeError: # pragma: no cover - return (), () - - # Get positional argument names (positional-only + positional-or-keyword) - args: tuple[str, ...] = code.co_varnames[: code.co_argcount] - - # Determine which args have defaults - kwargs: tuple[str, ...] - if defaults: - index = -len(defaults) - args, kwargs = args[:index], args[index:] - else: - kwargs = () - - # Strip implicit instance/class arg. - # Check if this looks like a method defined in a class by examining the - # qualname after the last "." segment (if any). A remaining dot - # means it's a class method (e.g. "MyClass.method" or - # "func..MyClass.method"), not just a nested function. - _tail = qualname.rsplit(".", maxsplit=1)[-1] - _is_class_method = "." in _tail - if args: - if is_bound or (_is_class_method and args[0] in _IMPLICIT_NAMES): - args = args[1:] - elif _is_class_method and legacy_noself and _tail not in _NOSELF_WARN_SUPPRESS: - warnings.warn( - f"{qualname} is a method but its first parameter" - f" {args[0]!r} is not 'self'." - f" Add 'self' as the first parameter or use @staticmethod." - f" This will become an error in a future version of pluggy.", - DeprecationWarning, - stacklevel=2, - ) - - return args, kwargs - - -@final -class HookRelay: - """Hook holder object for performing 1:N hook calls where N is the number - of registered plugins.""" - - __slots__ = ("__dict__",) - - def __init__(self) -> None: - """:meta private:""" - - if TYPE_CHECKING: - - def __getattr__(self, name: str) -> HookCaller: ... - - -# Historical name (pluggy<=1.2), kept for backward compatibility. -_HookRelay = HookRelay - - -_CallHistory: TypeAlias = list[ - tuple[Mapping[str, object], Callable[[Any], None] | None] -] - - -class HookCaller: - """A caller of all registered implementations of a hook specification.""" - - __slots__ = ( - "_call_history", - "_hookexec", - "_hookimpls", - "name", - "spec", - ) - - def __init__( - self, - name: str, - hook_execute: _HookExec, - specmodule_or_class: _Namespace | None = None, - spec_opts: HookspecOpts | None = None, - ) -> None: - """:meta private:""" - #: Name of the hook getting called. - self.name: Final = name - self._hookexec: Final = hook_execute - # The hookimpls list. The caller iterates it *in reverse*. Format: - # 1. trylast nonwrappers - # 2. nonwrappers - # 3. tryfirst nonwrappers - # 4. trylast wrappers - # 5. wrappers - # 6. tryfirst wrappers - self._hookimpls: Final[list[HookImpl]] = [] - self._call_history: _CallHistory | None = None - # TODO: Document, or make private. - self.spec: HookSpec | None = None - if specmodule_or_class is not None: - assert spec_opts is not None - self.set_specification(specmodule_or_class, spec_opts) - - # TODO: Document, or make private. - def has_spec(self) -> bool: - return self.spec is not None - - # TODO: Document, or make private. - def set_specification( - self, - specmodule_or_class: _Namespace, - spec_opts: HookspecOpts, - ) -> None: - if self.spec is not None: - raise ValueError( - f"Hook {self.spec.name!r} is already registered " - f"within namespace {self.spec.namespace}" - ) - self.spec = HookSpec(specmodule_or_class, self.name, spec_opts) - if spec_opts.get("historic"): - self._call_history = [] - - def is_historic(self) -> bool: - """Whether this caller is :ref:`historic `.""" - return self._call_history is not None - - def _remove_plugin(self, plugin: _Plugin) -> None: - """Remove all hook implementations registered by the given plugin.""" - remaining = [impl for impl in self._hookimpls if impl.plugin != plugin] - if len(remaining) == len(self._hookimpls): - raise ValueError(f"plugin {plugin!r} not found") - self._hookimpls[:] = remaining - - def get_hookimpls(self) -> list[HookImpl]: - """Get all registered hook implementations for this hook.""" - return self._hookimpls.copy() - - def _add_hookimpl(self, hookimpl: HookImpl) -> None: - """Add an implementation to the callback chain.""" - for i, method in enumerate(self._hookimpls): - if method.hookwrapper or method.wrapper: - splitpoint = i - break - else: - splitpoint = len(self._hookimpls) - if hookimpl.hookwrapper or hookimpl.wrapper: - start, end = splitpoint, len(self._hookimpls) - else: - start, end = 0, splitpoint - - if hookimpl.trylast: - self._hookimpls.insert(start, hookimpl) - elif hookimpl.tryfirst: - self._hookimpls.insert(end, hookimpl) - else: - # find last non-tryfirst method - i = end - 1 - while i >= start and self._hookimpls[i].tryfirst: - i -= 1 - self._hookimpls.insert(i + 1, hookimpl) - - def __repr__(self) -> str: - return f"" - - def _apply_defaults(self, kwargs: Mapping[str, object]) -> Mapping[str, object]: - if self.spec is None or not self.spec.kwargdefaults: - return kwargs - return {**self.spec.kwargdefaults, **kwargs} - - def _verify_all_args_are_provided(self, kwargs: Mapping[str, object]) -> None: - # This is written to avoid expensive operations when not needed. - if self.spec: - for argname in self.spec.argnames: - if argname not in kwargs: - notincall = ", ".join( - repr(argname) - for argname in self.spec.argnames - # Avoid self.spec.argnames - kwargs.keys() - # it doesn't preserve order. - if argname not in kwargs - ) - warnings.warn( - f"Argument(s) {notincall} which are declared in the hookspec " - "cannot be found in this hook call", - # 3, not 2: the warning is raised in this helper, which - # is called by __call__/call_historic/call_extra, which - # are called by the code making the hook call. - stacklevel=3, - ) - break - - def __call__(self, **kwargs: object) -> Any: - """Call the hook. - - Only accepts keyword arguments, which should match the hook - specification. - - Returns the result(s) of calling all registered plugins, see - :ref:`calling`. - """ - assert not self.is_historic(), ( - "Cannot directly call a historic hook - use call_historic instead." - ) - call_kwargs = self._apply_defaults(kwargs) - self._verify_all_args_are_provided(call_kwargs) - firstresult = self.spec.opts.get("firstresult", False) if self.spec else False - # Copy because plugins may register other plugins during iteration (#438). - return self._hookexec( - self.name, self._hookimpls.copy(), call_kwargs, firstresult - ) - - def call_historic( - self, - result_callback: Callable[[Any], None] | None = None, - kwargs: Mapping[str, object] | None = None, - ) -> None: - """Call the hook with given ``kwargs`` for all registered plugins and - for all plugins which will be registered afterwards, see - :ref:`historic`. - - :param result_callback: - If provided, will be called for each non-``None`` result obtained - from a hook implementation. - """ - assert self._call_history is not None - kwargs = kwargs or {} - kwargs = self._apply_defaults(kwargs) - self._verify_all_args_are_provided(kwargs) - self._call_history.append((kwargs, result_callback)) - # Historizing hooks don't return results. - # Remember firstresult isn't compatible with historic. - # Copy because plugins may register other plugins during iteration (#438). - res = self._hookexec(self.name, self._hookimpls.copy(), kwargs, False) - if result_callback is None: - return - if isinstance(res, list): - for x in res: - result_callback(x) - - def call_extra( - self, methods: Sequence[Callable[..., object]], kwargs: Mapping[str, object] - ) -> Any: - """Call the hook with some additional temporarily participating - methods using the specified ``kwargs`` as call parameters, see - :ref:`call_extra`.""" - assert not self.is_historic(), ( - "Cannot directly call a historic hook - use call_historic instead." - ) - kwargs = self._apply_defaults(kwargs) - self._verify_all_args_are_provided(kwargs) - opts: HookimplOpts = { - "wrapper": False, - "hookwrapper": False, - "optionalhook": False, - "trylast": False, - "tryfirst": False, - "specname": None, - } - hookimpls = self._hookimpls.copy() - for method in methods: - hookimpl = HookImpl(None, "", method, opts) - # Find last non-tryfirst nonwrapper method. - i = len(hookimpls) - 1 - while i >= 0 and ( - # Skip wrappers. - (hookimpls[i].hookwrapper or hookimpls[i].wrapper) - # Skip tryfirst nonwrappers. - or hookimpls[i].tryfirst - ): - i -= 1 - hookimpls.insert(i + 1, hookimpl) - firstresult = self.spec.opts.get("firstresult", False) if self.spec else False - return self._hookexec(self.name, hookimpls, kwargs, firstresult) - - def _maybe_apply_history(self, method: HookImpl) -> None: - """Apply call history to a new hookimpl if it is marked as historic.""" - if self.is_historic(): - assert self._call_history is not None - for kwargs, result_callback in self._call_history: - res = self._hookexec(self.name, [method], kwargs, False) - if res and result_callback is not None: - # XXX: remember firstresult isn't compat with historic - assert isinstance(res, list) - result_callback(res[0]) - - -# Historical name (pluggy<=1.2), kept for backward compatibility. -_HookCaller = HookCaller - - -class _SubsetHookCaller(HookCaller): - """A proxy to another HookCaller which manages calls to all registered - plugins except the ones from remove_plugins.""" - - # This class is unusual: in inhertits from `HookCaller` so all of - # the *code* runs in the class, but it delegates all underlying *data* - # to the original HookCaller. - # `subset_hook_caller` used to be implemented by creating a full-fledged - # HookCaller, copying all hookimpls from the original. This had problems - # with memory leaks (#346) and historic calls (#347), which make a proxy - # approach better. - # An alternative implementation is to use a `_getattr__`/`__getattribute__` - # proxy, however that adds more overhead and is more tricky to implement. - - __slots__ = ( - "_orig", - "_remove_plugins", - ) - - def __init__(self, orig: HookCaller, remove_plugins: AbstractSet[_Plugin]) -> None: - self._orig = orig - self._remove_plugins = remove_plugins - self.name = orig.name # type: ignore[misc] - self._hookexec = orig._hookexec # type: ignore[misc] - - @property # type: ignore[misc] - def _hookimpls(self) -> list[HookImpl]: - return [ - impl - for impl in self._orig._hookimpls - if impl.plugin not in self._remove_plugins - ] - - @property - def spec(self) -> HookSpec | None: # type: ignore[override] - return self._orig.spec - - @property - def _call_history(self) -> _CallHistory | None: # type: ignore[override] - return self._orig._call_history - - def __repr__(self) -> str: - return f"<_SubsetHookCaller {self.name!r}>" - - -@final -class HookImpl: - """A hook implementation in a :class:`HookCaller`.""" - - __slots__ = ( - "argnames", - "function", - "hookwrapper", - "kwargnames", - "optionalhook", - "opts", - "plugin", - "plugin_name", - "tryfirst", - "trylast", - "wrapper", - ) - - def __init__( - self, - plugin: _Plugin, - plugin_name: str, - function: _HookImplFunction[object], - hook_impl_opts: HookimplOpts, - ) -> None: - """:meta private:""" - #: The hook implementation function. - self.function: Final = function - argnames, kwargnames = varnames(self.function) - #: The positional parameter names of ``function```. - self.argnames: Final = argnames - #: The keyword parameter names of ``function```. - self.kwargnames: Final = kwargnames - #: The plugin which defined this hook implementation. - self.plugin: Final = plugin - #: The :class:`HookimplOpts` used to configure this hook implementation. - self.opts: Final = hook_impl_opts - #: The name of the plugin which defined this hook implementation. - self.plugin_name: Final = plugin_name - #: Whether the hook implementation is a :ref:`wrapper `. - self.wrapper: Final = hook_impl_opts["wrapper"] - #: Whether the hook implementation is an :ref:`old-style wrapper - #: `. - self.hookwrapper: Final = hook_impl_opts["hookwrapper"] - #: Whether validation against a hook specification is :ref:`optional - #: `. - self.optionalhook: Final = hook_impl_opts["optionalhook"] - #: Whether to try to order this hook implementation :ref:`first - #: `. - self.tryfirst: Final = hook_impl_opts["tryfirst"] - #: Whether to try to order this hook implementation :ref:`last - #: `. - self.trylast: Final = hook_impl_opts["trylast"] - - def __repr__(self) -> str: - return f"" - - -@final -class HookSpec: - __slots__ = ( - "argnames", - "function", - "kwargdefaults", - "kwargnames", - "name", - "namespace", - "opts", - "warn_on_impl", - "warn_on_impl_args", - ) - - def __init__(self, namespace: _Namespace, name: str, opts: HookspecOpts) -> None: - self.namespace = namespace - self.name = name - self.function: Callable[..., object] = getattr(namespace, name) - legacy_noself = inspect.isclass(namespace) and not isinstance( - inspect.getattr_static(namespace, name), staticmethod - ) - self.argnames, self.kwargnames = varnames( - self.function, legacy_noself=legacy_noself - ) - defaults = inspect.unwrap(self.function).__defaults__ - self.kwargdefaults = dict(zip(self.kwargnames, defaults or ())) - self.opts = opts - self.warn_on_impl = opts.get("warn_on_impl") - self.warn_on_impl_args = opts.get("warn_on_impl_args") diff --git a/src/pluggy/_impl.py b/src/pluggy/_impl.py new file mode 100644 index 00000000..4f82ebe5 --- /dev/null +++ b/src/pluggy/_impl.py @@ -0,0 +1,80 @@ +""" +Hook implementation representation. +""" + +from __future__ import annotations + +from collections.abc import Callable +from collections.abc import Generator +from typing import Final +from typing import final +from typing import TypeAlias +from typing import TypeVar + +from ._config import HookimplOpts +from ._decorators import varnames +from ._result import Result + + +_T = TypeVar("_T") + +_Plugin: TypeAlias = object +_HookImplFunction: TypeAlias = Callable[..., _T | Generator[None, Result[_T], None]] + + +@final +class HookImpl: + """A hook implementation in a :class:`HookCaller`.""" + + __slots__ = ( + "argnames", + "function", + "hookwrapper", + "kwargnames", + "optionalhook", + "opts", + "plugin", + "plugin_name", + "tryfirst", + "trylast", + "wrapper", + ) + + def __init__( + self, + plugin: _Plugin, + plugin_name: str, + function: _HookImplFunction[object], + hook_impl_opts: HookimplOpts, + ) -> None: + """:meta private:""" + #: The hook implementation function. + self.function: Final = function + argnames, kwargnames = varnames(self.function) + #: The positional parameter names of ``function```. + self.argnames: Final = argnames + #: The keyword parameter names of ``function```. + self.kwargnames: Final = kwargnames + #: The plugin which defined this hook implementation. + self.plugin: Final = plugin + #: The :class:`HookimplOpts` used to configure this hook implementation. + self.opts: Final = hook_impl_opts + #: The name of the plugin which defined this hook implementation. + self.plugin_name: Final = plugin_name + #: Whether the hook implementation is a :ref:`wrapper `. + self.wrapper: Final = hook_impl_opts["wrapper"] + #: Whether the hook implementation is an :ref:`old-style wrapper + #: `. + self.hookwrapper: Final = hook_impl_opts["hookwrapper"] + #: Whether validation against a hook specification is :ref:`optional + #: `. + self.optionalhook: Final = hook_impl_opts["optionalhook"] + #: Whether to try to order this hook implementation :ref:`first + #: `. + self.tryfirst: Final = hook_impl_opts["tryfirst"] + #: Whether to try to order this hook implementation :ref:`last + #: `. + self.trylast: Final = hook_impl_opts["trylast"] + + def __repr__(self) -> str: + return f""