pluggy plugin system with hookspecs, firstresult hooks and wrappers: working versions (tested 2026-10-18)

pluggy==1.7.0
pytest==9.1.1

import pluggy

project_name = "provenpin"
hookspec = pluggy.HookspecMarker(project_name)
hookimpl = pluggy.HookimplMarker(project_name)


class ProcessorSpec:
    @hookspec
    def enrich(self, record: dict) -> None:
        """Mutate the record in place. Every plugin gets a chance to run."""

    @hookspec(firstresult=True)
    def score(self, record: dict) -> int | None:
        """Return the first non-None score, or None when no plugin scores it."""

    @hookspec
    def audit(self, event: str, payload: dict) -> None:
        """Side-effect hook. All plugins and wrappers are called."""

from hookspecs import hookimpl


class Enricher:
    def __init__(self, tag: str):
        self.tag = tag

    @hookimpl
    def enrich(self, record: dict) -> None:
        record.setdefault("order", []).append(self.tag)


class FirstEnricher:
    @hookimpl(tryfirst=True)
    def enrich(self, record: dict) -> None:
        record.setdefault("order", []).append("first")


class LastEnricher:
    @hookimpl(trylast=True)
    def enrich(self, record: dict) -> None:
        record.setdefault("order", []).append("last")


class PriorityScorer:
    @hookimpl(tryfirst=True)
    def score(self, record: dict) -> int | None:
        return record.get("priority")


class FallbackScorer:
    @hookimpl(trylast=True)
    def score(self, record: dict) -> int | None:
        return 1


class ScoreClamp:
    @hookimpl(wrapper=True)
    def score(self, record: dict):
        raw = yield
        return min(raw or 0, 50)


class Outer:
    def __init__(self, log: list):
        self.log = log

    @hookimpl(wrapper=True, tryfirst=True)
    def audit(self, event: str, payload: dict):
        self.log.append("outer-before")
        try:
            return (yield)
        finally:
            self.log.append("outer-after")


class Inner:
    def __init__(self, log: list):
        self.log = log

    @hookimpl(wrapper=True, trylast=True)
    def audit(self, event: str, payload: dict):
        self.log.append("inner-before")
        try:
            return (yield)
        finally:
            self.log.append("inner-after")


class AuditLog:
    def __init__(self, log: list):
        self.log = log

    @hookimpl
    def audit(self, event: str, payload: dict) -> None:
        self.log.append(f"impl:{event}")

import pluggy

from hookspecs import ProcessorSpec, project_name


def build_pm() -> pluggy.PluginManager:
    pm = pluggy.PluginManager(project_name)
    pm.add_hookspecs(ProcessorSpec)
    return pm


def process(pm: pluggy.PluginManager, record: dict) -> dict:
    pm.hook.enrich(record=record)
    record["score"] = pm.hook.score(record=record)
    pm.hook.audit(event="processed", payload=record)
    return record

import pluggy
import pytest

from core import build_pm, process
from hookspecs import hookimpl
from plugins import (
    AuditLog,
    Enricher,
    FallbackScorer,
    FirstEnricher,
    Inner,
    LastEnricher,
    Outer,
    PriorityScorer,
    ScoreClamp,
)


def test_all_impls_run_in_lifo_order_unless_tryfirst_or_trylast_says_otherwise():
    pm = build_pm()
    pm.register(Enricher("a"))
    pm.register(Enricher("b"))
    pm.register(FirstEnricher())
    pm.register(LastEnricher())
    record = process(pm, {})
    assert record["order"] == ["first", "b", "a", "last"]


def test_firstresult_takes_first_non_none_and_falls_through_to_trylast():
    pm = build_pm()
    pm.register(PriorityScorer())
    pm.register(FallbackScorer())
    assert process(pm, {"priority": 7})["score"] == 7
    assert process(pm, {})["score"] == 1
    assert process(build_pm(), {})["score"] is None


def test_wrapper_can_transform_a_firstresult_value():
    pm = build_pm()
    pm.register(ScoreClamp())
    pm.register(PriorityScorer())
    assert process(pm, {"priority": 100})["score"] == 50


def test_tryfirst_wrapper_is_outermost_and_trylast_wrapper_is_innermost():
    pm = build_pm()
    log = []
    pm.register(AuditLog(log))
    pm.register(Inner(log))
    pm.register(Outer(log))
    process(pm, {})
    assert log == [
        "outer-before",
        "inner-before",
        "impl:processed",
        "inner-after",
        "outer-after",
    ]


def test_unknown_hookimpl_registers_silently_until_check_pending():
    class Rogue:
        @hookimpl
        def does_not_exist(self):
            return 1

    pm = build_pm()
    pm.register(Rogue())
    with pytest.raises(pluggy.PluginValidationError):
        pm.check_pending()


def test_optionalhook_lets_a_plugin_ahead_of_its_hookspec_register_cleanly():
    class Future:
        @hookimpl(optionalhook=True)
        def does_not_exist_yet(self):
            return 1

    pm = build_pm()
    pm.register(Future(), "future")
    assert pm.get_plugin("future").__class__ is Future

[pytest]
pythonpath = .