From a097fb3ab11aada6cec637350cfcdcb51885b5ae Mon Sep 17 00:00:00 2001 From: Ronny Pfannschmidt Date: Sat, 29 Nov 2025 00:16:19 +0100 Subject: [PATCH] Introduce immutable hook option classes for type safety MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hook implementation and specification options are now represented by immutable classes (HookimplOptions, HookspecOptions) internally. This provides stronger type guarantees and prevents accidental mutation of option data after hook registration. For backward compatibility with pytest's custom parsing methods, the public parse_hookimpl_opts and parse_hookspec_opts methods continue to return TypedDict representations, while new internal helpers handle conversion to the immutable classes. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/pluggy/_hooks.py | 255 ++++++++++++++++++++++++++++++------- src/pluggy/_manager.py | 81 ++++++++---- testing/test_hookcaller.py | 8 +- 3 files changed, 266 insertions(+), 78 deletions(-) diff --git a/src/pluggy/_hooks.py b/src/pluggy/_hooks.py index 7fde78c9..9327b0bb 100644 --- a/src/pluggy/_hooks.py +++ b/src/pluggy/_hooks.py @@ -74,6 +74,185 @@ class HookimplOpts(TypedDict): specname: str | None +@final +class HookspecOptions: + """Immutable hook specification options (internal). + + This is the internal representation used by pluggy. The TypedDict + :class:`HookspecOpts` is kept for backward compatibility with pytest. + """ + + __slots__ = ("firstresult", "historic", "warn_on_impl", "warn_on_impl_args") + + #: Whether to stop at the first non-None result. + firstresult: Final[bool] # type: ignore[misc] + #: Whether this hook is :ref:`historic `. + historic: Final[bool] # type: ignore[misc] + #: A warning to emit when a hook implementation is registered. + warn_on_impl: Final[Warning | None] # type: ignore[misc] + #: Warnings to emit for specific arguments when a hook implementation is registered. + warn_on_impl_args: Final[Mapping[str, Warning] | None] # type: ignore[misc] + + def __init__( + self, + firstresult: bool = False, + historic: bool = False, + warn_on_impl: Warning | None = None, + warn_on_impl_args: Mapping[str, Warning] | None = None, + ) -> None: + object.__setattr__(self, "firstresult", firstresult) + object.__setattr__(self, "historic", historic) + object.__setattr__(self, "warn_on_impl", warn_on_impl) + object.__setattr__(self, "warn_on_impl_args", warn_on_impl_args) + + def __setattr__(self, name: str, value: object) -> None: + raise AttributeError("HookspecOptions is immutable") + + def __delattr__(self, name: str) -> None: + raise AttributeError("HookspecOptions is immutable") + + def __repr__(self) -> str: + return ( + f"HookspecOptions(firstresult={self.firstresult!r}, " + f"historic={self.historic!r}, " + f"warn_on_impl={self.warn_on_impl!r}, " + f"warn_on_impl_args={self.warn_on_impl_args!r})" + ) + + def __eq__(self, other: object) -> bool: + if not isinstance(other, HookspecOptions): + return NotImplemented + return ( + self.firstresult == other.firstresult + and self.historic == other.historic + and self.warn_on_impl == other.warn_on_impl + and self.warn_on_impl_args == other.warn_on_impl_args + ) + + def __hash__(self) -> int: + return hash( + ( + self.firstresult, + self.historic, + self.warn_on_impl, + # warn_on_impl_args is a Mapping, convert to hashable + tuple(self.warn_on_impl_args.items()) + if self.warn_on_impl_args + else None, + ) + ) + + @classmethod + def from_opts(cls, opts: HookspecOpts) -> HookspecOptions: + """Create from a HookspecOpts TypedDict (for backward compatibility).""" + return cls( + firstresult=opts.get("firstresult", False), + historic=opts.get("historic", False), + warn_on_impl=opts.get("warn_on_impl"), + warn_on_impl_args=opts.get("warn_on_impl_args"), + ) + + +@final +class HookimplOptions: + """Immutable hook implementation options (internal). + + This is the internal representation used by pluggy. The TypedDict + :class:`HookimplOpts` is kept for backward compatibility with pytest. + """ + + __slots__ = ( + "wrapper", + "hookwrapper", + "optionalhook", + "tryfirst", + "trylast", + "specname", + ) + + #: Whether this is a :ref:`wrapper `. + wrapper: Final[bool] # type: ignore[misc] + #: Whether this is an :ref:`old-style wrapper `. + hookwrapper: Final[bool] # type: ignore[misc] + #: Whether validation against a hook specification is :ref:`optional + #: `. + optionalhook: Final[bool] # type: ignore[misc] + #: Whether to try to order this hook implementation :ref:`first `. + tryfirst: Final[bool] # type: ignore[misc] + #: Whether to try to order this hook implementation :ref:`last `. + trylast: Final[bool] # type: ignore[misc] + #: The name of the hook specification to match, see :ref:`specname`. + specname: Final[str | None] # type: ignore[misc] + + def __init__( + self, + wrapper: bool = False, + hookwrapper: bool = False, + optionalhook: bool = False, + tryfirst: bool = False, + trylast: bool = False, + specname: str | None = None, + ) -> None: + object.__setattr__(self, "wrapper", wrapper) + object.__setattr__(self, "hookwrapper", hookwrapper) + object.__setattr__(self, "optionalhook", optionalhook) + object.__setattr__(self, "tryfirst", tryfirst) + object.__setattr__(self, "trylast", trylast) + object.__setattr__(self, "specname", specname) + + def __setattr__(self, name: str, value: object) -> None: + raise AttributeError("HookimplOptions is immutable") + + def __delattr__(self, name: str) -> None: + raise AttributeError("HookimplOptions is immutable") + + def __repr__(self) -> str: + return ( + f"HookimplOptions(wrapper={self.wrapper!r}, " + f"hookwrapper={self.hookwrapper!r}, " + f"optionalhook={self.optionalhook!r}, " + f"tryfirst={self.tryfirst!r}, " + f"trylast={self.trylast!r}, " + f"specname={self.specname!r})" + ) + + def __eq__(self, other: object) -> bool: + if not isinstance(other, HookimplOptions): + return NotImplemented + return ( + self.wrapper == other.wrapper + and self.hookwrapper == other.hookwrapper + and self.optionalhook == other.optionalhook + and self.tryfirst == other.tryfirst + and self.trylast == other.trylast + and self.specname == other.specname + ) + + def __hash__(self) -> int: + return hash( + ( + self.wrapper, + self.hookwrapper, + self.optionalhook, + self.tryfirst, + self.trylast, + self.specname, + ) + ) + + @classmethod + def from_opts(cls, opts: HookimplOpts) -> HookimplOptions: + """Create from a HookimplOpts TypedDict (for backward compatibility).""" + return cls( + wrapper=opts.get("wrapper", False), + hookwrapper=opts.get("hookwrapper", False), + optionalhook=opts.get("optionalhook", False), + tryfirst=opts.get("tryfirst", False), + trylast=opts.get("trylast", False), + specname=opts.get("specname"), + ) + + @final class HookspecMarker: """Decorator for marking functions as hook specifications. @@ -146,12 +325,12 @@ def __call__( # noqa: F811 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, - } + opts = HookspecOptions( + 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 @@ -261,14 +440,14 @@ def __call__( # noqa: F811 """ def setattr_hookimpl_opts(func: _F) -> _F: - opts: HookimplOpts = { - "wrapper": wrapper, - "hookwrapper": hookwrapper, - "optionalhook": optionalhook, - "tryfirst": tryfirst, - "trylast": trylast, - "specname": specname, - } + opts = HookimplOptions( + wrapper=wrapper, + hookwrapper=hookwrapper, + optionalhook=optionalhook, + tryfirst=tryfirst, + trylast=trylast, + specname=specname, + ) setattr(func, self.project_name + "_impl", opts) return func @@ -278,15 +457,6 @@ def setattr_hookimpl_opts(func: _F) -> _F: 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 = hasattr(sys, "pypy_version_info") @@ -395,7 +565,7 @@ def __init__( name: str, hook_execute: _HookExec, specmodule_or_class: _Namespace | None = None, - spec_opts: HookspecOpts | None = None, + spec_opts: HookspecOptions | None = None, ) -> None: """:meta private:""" #: Name of the hook getting called. @@ -424,7 +594,7 @@ def has_spec(self) -> bool: def set_specification( self, specmodule_or_class: _Namespace, - spec_opts: HookspecOpts, + spec_opts: HookspecOptions, ) -> None: if self.spec is not None: raise ValueError( @@ -432,7 +602,7 @@ def set_specification( f"within namespace {self.spec.namespace}" ) self.spec = HookSpec(specmodule_or_class, self.name, spec_opts) - if spec_opts.get("historic"): + if spec_opts.historic: self._call_history = [] def is_historic(self) -> bool: @@ -509,7 +679,7 @@ def __call__(self, **kwargs: object) -> Any: "Cannot directly call a historic hook - use call_historic instead." ) self._verify_all_args_are_provided(kwargs) - firstresult = self.spec.opts.get("firstresult", False) if self.spec else False + firstresult = self.spec.opts.firstresult if self.spec else False # Copy because plugins may register other plugins during iteration (#438). return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult) @@ -550,14 +720,7 @@ def call_extra( "Cannot directly call a historic hook - use call_historic instead." ) self._verify_all_args_are_provided(kwargs) - opts: HookimplOpts = { - "wrapper": False, - "hookwrapper": False, - "optionalhook": False, - "trylast": False, - "tryfirst": False, - "specname": None, - } + opts = HookimplOptions() hookimpls = self._hookimpls.copy() for method in methods: hookimpl = HookImpl(None, "", method, opts) @@ -571,7 +734,7 @@ def call_extra( ): i -= 1 hookimpls.insert(i + 1, hookimpl) - firstresult = self.spec.opts.get("firstresult", False) if self.spec else False + firstresult = self.spec.opts.firstresult if self.spec else False return self._hookexec(self.name, hookimpls, kwargs, firstresult) def _maybe_apply_history(self, method: HookImpl) -> None: @@ -658,7 +821,7 @@ def __init__( plugin: _Plugin, plugin_name: str, function: _HookImplFunction[object], - hook_impl_opts: HookimplOpts, + hook_impl_opts: HookimplOptions, ) -> None: """:meta private:""" #: The hook implementation function. @@ -670,24 +833,24 @@ def __init__( self.kwargnames: Final = kwargnames #: The plugin which defined this hook implementation. self.plugin: Final = plugin - #: The :class:`HookimplOpts` used to configure this hook implementation. + #: The :class:`HookimplOptions` 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"] + self.wrapper: Final = hook_impl_opts.wrapper #: Whether the hook implementation is an :ref:`old-style wrapper #: `. - self.hookwrapper: Final = hook_impl_opts["hookwrapper"] + self.hookwrapper: Final = hook_impl_opts.hookwrapper #: Whether validation against a hook specification is :ref:`optional #: `. - self.optionalhook: Final = hook_impl_opts["optionalhook"] + self.optionalhook: Final = hook_impl_opts.optionalhook #: Whether to try to order this hook implementation :ref:`first #: `. - self.tryfirst: Final = hook_impl_opts["tryfirst"] + self.tryfirst: Final = hook_impl_opts.tryfirst #: Whether to try to order this hook implementation :ref:`last #: `. - self.trylast: Final = hook_impl_opts["trylast"] + self.trylast: Final = hook_impl_opts.trylast def __repr__(self) -> str: return f"" @@ -706,11 +869,11 @@ class HookSpec: "warn_on_impl_args", ) - def __init__(self, namespace: _Namespace, name: str, opts: HookspecOpts) -> None: + def __init__(self, namespace: _Namespace, name: str, opts: HookspecOptions) -> None: self.namespace = namespace self.function: Callable[..., object] = getattr(namespace, name) self.name = name self.argnames, self.kwargnames = varnames(self.function) self.opts = opts - self.warn_on_impl = opts.get("warn_on_impl") - self.warn_on_impl_args = opts.get("warn_on_impl_args") + self.warn_on_impl = opts.warn_on_impl + self.warn_on_impl_args = opts.warn_on_impl_args diff --git a/src/pluggy/_manager.py b/src/pluggy/_manager.py index 1b994f25..fe08c8d6 100644 --- a/src/pluggy/_manager.py +++ b/src/pluggy/_manager.py @@ -21,10 +21,11 @@ from ._hooks import _SubsetHookCaller from ._hooks import HookCaller from ._hooks import HookImpl +from ._hooks import HookimplOptions from ._hooks import HookimplOpts from ._hooks import HookRelay +from ._hooks import HookspecOptions from ._hooks import HookspecOpts -from ._hooks import normalize_hookimpl_opts from ._result import Result @@ -157,12 +158,11 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None: # register matching hook implementations of the plugin for name in dir(plugin): - hookimpl_opts = self.parse_hookimpl_opts(plugin, name) + hookimpl_opts = self._parse_hookimpl_opts(plugin, name) if hookimpl_opts is not None: - normalize_hookimpl_opts(hookimpl_opts) method: _HookImplFunction[object] = getattr(plugin, name) hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts) - name = hookimpl_opts.get("specname") or name + name = hookimpl_opts.specname or name hook: HookCaller | None = getattr(self.hook, name, None) if hook is None: hook = HookCaller(name, self._hookexec) @@ -173,30 +173,43 @@ def register(self, plugin: _Plugin, name: str | None = None) -> str | None: hook._add_hookimpl(hookimpl) return plugin_name + def _parse_hookimpl_opts( + self, plugin: _Plugin, name: str + ) -> HookimplOptions | None: + """Internal: parse hookimpl opts and convert to HookimplOptions.""" + try: + method: object = getattr(plugin, name) + except Exception: + return None + if not inspect.isroutine(method): + return None + try: + res: HookimplOptions | None = getattr( + method, self.project_name + "_impl", None + ) + except Exception: + return None + if res is not None: + return res + # Check for legacy dict from overridden parse_hookimpl_opts + legacy = self.parse_hookimpl_opts(plugin, name) + if legacy is not None: + return HookimplOptions.from_opts(legacy) + return None + def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None: """Try to obtain a hook implementation from an item with the given name in the given plugin which is being searched for hook impls. :returns: - The parsed hookimpl options, or None to skip the given item. + The parsed hookimpl options as a dict, or None to skip the given item. This method can be overridden by ``PluginManager`` subclasses to - customize how hook implementation are picked up. By default, returns the - options for items decorated with :class:`HookimplMarker`. + customize how hook implementations are picked up. By default, returns + None. Override this method to return a :class:`HookimplOpts` dict for + items that should be treated as hook implementations. """ - method: object = getattr(plugin, name) - if not inspect.isroutine(method): - return None - try: - res: HookimplOpts | None = getattr( - method, self.project_name + "_impl", None - ) - except Exception: # pragma: no cover - res = {} # type: ignore[assignment] #pragma: no cover - if res is not None and not isinstance(res, dict): - # false positive - res = None # type:ignore[unreachable] #pragma: no cover - return res + return None def unregister( self, plugin: _Plugin | None = None, name: str | None = None @@ -257,7 +270,7 @@ def add_hookspecs(self, module_or_class: _Namespace) -> None: """ names = [] for name in dir(module_or_class): - spec_opts = self.parse_hookspec_opts(module_or_class, name) + spec_opts = self._parse_hookspec_opts(module_or_class, name) if spec_opts is not None: hc: HookCaller | None = getattr(self.hook, name, None) if hc is None: @@ -275,6 +288,20 @@ def add_hookspecs(self, module_or_class: _Namespace) -> None: f"did not find any {self.project_name!r} hooks in {module_or_class!r}" ) + def _parse_hookspec_opts( + self, module_or_class: _Namespace, name: str + ) -> HookspecOptions | None: + """Internal: parse hookspec opts and convert to HookspecOptions.""" + method = getattr(module_or_class, name) + res: HookspecOptions | None = getattr(method, self.project_name + "_spec", None) + if res is not None: + return res + # Check for legacy dict from overridden parse_hookspec_opts + legacy = self.parse_hookspec_opts(module_or_class, name) + if legacy is not None: + return HookspecOptions.from_opts(legacy) + return None + def parse_hookspec_opts( self, module_or_class: _Namespace, name: str ) -> HookspecOpts | None: @@ -282,16 +309,14 @@ def parse_hookspec_opts( in the given module or class which is being searched for hook specs. :returns: - The parsed hookspec options for defining a hook, or None to skip the - given item. + The parsed hookspec options as a dict, or None to skip the given item. This method can be overridden by ``PluginManager`` subclasses to - customize how hook specifications are picked up. By default, returns the - options for items decorated with :class:`HookspecMarker`. + customize how hook specifications are picked up. By default, returns + None. Override this method to return a :class:`HookspecOpts` dict for + items that should be treated as hook specifications. """ - method = getattr(module_or_class, name) - opts: HookspecOpts | None = getattr(method, self.project_name + "_spec", None) - return opts + return None def get_plugins(self) -> set[Any]: """Return a set of all registered plugin objects.""" diff --git a/testing/test_hookcaller.py b/testing/test_hookcaller.py index 5c42f4b6..8a7982f6 100644 --- a/testing/test_hookcaller.py +++ b/testing/test_hookcaller.py @@ -317,11 +317,11 @@ def he_myhook3(arg1) -> None: pm.add_hookspecs(HookSpec) assert pm.hook.he_myhook1.spec is not None - assert not pm.hook.he_myhook1.spec.opts["firstresult"] + assert not pm.hook.he_myhook1.spec.opts.firstresult assert pm.hook.he_myhook2.spec is not None - assert pm.hook.he_myhook2.spec.opts["firstresult"] + assert pm.hook.he_myhook2.spec.opts.firstresult assert pm.hook.he_myhook3.spec is not None - assert not pm.hook.he_myhook3.spec.opts["firstresult"] + assert not pm.hook.he_myhook3.spec.opts.firstresult @pytest.mark.parametrize("name", ["hookwrapper", "optionalhook", "tryfirst", "trylast"]) @@ -332,7 +332,7 @@ def he_myhook1(arg1) -> None: pass if val: - assert he_myhook1.example_impl.get(name) + assert getattr(he_myhook1.example_impl, name) else: assert not hasattr(he_myhook1, name)