Skip to content

API Reference

Generated from the source docstrings. The public API is re-exported from the top-level pluginkit package.

Public API

pluginkit

pluginkit: a small, strictly-typed, generics-first plugin framework for Python 3.13+.

Unlike untyped hook systems, pluginkit derives a hook call's return type from its spec: pm.caller(spec) hands back a caller whose result is list[R] (collecting), R | None (firstresult), or R (pipeline) - checked, not asserted.

Public API:

  • :class:ExtensionPoint / :class:Extension - decorators that declare extension points and the extensions that fulfil them. @extension_point brands the declaration by dispatch mode.
  • :class:ExtensionPointOpts / :class:ExtensionOpts - the option records the markers stamp.
  • :class:PluginManager - registers plugins and dispatches calls; caller(spec) returns a typed caller.
  • :class:CollectingSpec / :class:FirstResultSpec / :class:PipelineSpec - branded spec types, and :class:CollectingCaller / :class:FirstResultCaller / :class:PipelineCaller (and the Async* variants) - the typed callers.
  • :class:HookRelay / :class:HookCaller / :class:HookImpl - the dispatch internals.
  • :class:PluginValidationError - raised when a plugin is invalid.

Classes

AsyncCollectingCaller dataclass

Bases: AsyncHookCaller

A collecting async hook's typed caller: await a call to get list[R].

Source code in src/pluginkit/aio.py 
120
121
122
123
124
125
class AsyncCollectingCaller[**P, R](AsyncHookCaller):
    """A collecting async hook's typed caller: `await` a call to get `list[R]`."""

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
        """Await the collecting hook, returning each impl's result as `list[R]`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller
Methods:
__call__(*args, **kwargs) async

Await the collecting hook, returning each impl's result as list[R].

Source code in src/pluginkit/aio.py 
123
124
125
async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
    """Await the collecting hook, returning each impl's result as `list[R]`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller

AsyncFirstResultCaller dataclass

Bases: AsyncHookCaller

A firstresult async hook's typed caller: await a call to get R | None.

Source code in src/pluginkit/aio.py 
128
129
130
131
132
133
class AsyncFirstResultCaller[**P, R](AsyncHookCaller):
    """A firstresult async hook's typed caller: `await` a call to get `R | None`."""

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
        """Await the firstresult hook, returning the first non-None `R` or `None`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller
Methods:
__call__(*args, **kwargs) async

Await the firstresult hook, returning the first non-None R or None.

Source code in src/pluginkit/aio.py 
131
132
133
async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
    """Await the firstresult hook, returning the first non-None `R` or `None`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller

AsyncHookCaller dataclass

Bases: HookCaller

A HookCaller whose calls are coroutines that await async implementations.

Source code in src/pluginkit/aio.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class AsyncHookCaller(HookCaller):
    """A HookCaller whose calls are coroutines that await async implementations."""

    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
        """Await the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return await self._execute_async(kwargs, self._nonwrappers)

    async def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
        """Await the hook with extra one-off implementations that are not registered."""
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(kwargs)
        combined = sorted(
            [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
        )
        return await self._execute_async(kwargs, combined)

    def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
        """Historic hooks are not supported by the async manager."""
        raise NotImplementedError("historic hooks are not supported by AsyncPluginManager")

    async def _execute_async(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Start async wrappers, run the impls, then unwind wrappers exception-safely."""
        started: list[AsyncGeneratorType[Any, Any]] = []
        try:
            for wrapper in self._wrappers:
                generator = wrapper.call(kwargs)
                if not isinstance(generator, AsyncGeneratorType):
                    raise TypeError(f"async wrapper {wrapper.plugin_name}.{self.name} must be an async generator")
                await generator.__anext__()  # advance to the yield
                started.append(generator)
            result = await self._core_async(kwargs, nonwrappers)
        except BaseException as exc:  # noqa: BLE001 - re-raised after wrappers observe it
            return await self._teardown_async(started, exc=exc)
        return await self._teardown_async(started, result=result)

    async def _core_async(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Apply the spec's dispatch strategy, awaiting any awaitable results."""
        if self.spec.pipeline:
            return await self._run_pipeline_async(kwargs, nonwrappers)
        results: list[Any] = []
        for impl in nonwrappers:
            outcome = await _maybe_await(impl.call(kwargs))
            if outcome is None:
                continue
            results.append(outcome)
            if self.spec.firstresult:
                break
        return (results[0] if results else None) if self.spec.firstresult else results

    async def _run_pipeline_async(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Thread the first argument through each impl, awaiting awaitable results."""
        param = self.params[0]
        value = kwargs[param]
        current = dict(kwargs)
        for impl in nonwrappers:
            current[param] = value
            outcome = await _maybe_await(impl.call(current))
            if outcome is not None:
                value = outcome
        return value

    async def _teardown_async(
        self, started: list[AsyncGeneratorType[Any, Any]], *, result: Any = _UNSET, exc: BaseException | None = None
    ) -> Any:
        """Resume each async wrapper in reverse so its teardown runs and it observes errors."""
        for generator in reversed(started):
            try:
                if exc is not None:
                    await generator.athrow(exc)
                else:
                    await generator.asend(result)
            except StopAsyncIteration:
                pass  # normal completion; async wrappers cannot replace the result
            except BaseException as new_exc:  # noqa: BLE001 - propagate the wrapper's error onward
                exc = new_exc
            else:
                # Double yield: capture the error but keep unwinding the remaining
                # wrappers so their teardown still runs; raised after the loop.
                await generator.aclose()
                exc = RuntimeError(f"async wrapper for {self.name!r} must yield exactly once")
        if exc is not None:
            raise exc
        return result
Methods:
__call__(*args, **kwargs) async

Await the hook: a list, a single value (firstresult), or the threaded value (pipeline).

Source code in src/pluginkit/aio.py 
32
33
34
35
36
37
async def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Await the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return await self._execute_async(kwargs, self._nonwrappers)
call_extra(functions, kwargs) async

Await the hook with extra one-off implementations that are not registered.

Source code in src/pluginkit/aio.py 
39
40
41
42
43
44
45
46
47
async def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
    """Await the hook with extra one-off implementations that are not registered."""
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(kwargs)
    combined = sorted(
        [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
    )
    return await self._execute_async(kwargs, combined)
call_historic(kwargs, result_callback=None)

Historic hooks are not supported by the async manager.

Source code in src/pluginkit/aio.py 
49
50
51
def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
    """Historic hooks are not supported by the async manager."""
    raise NotImplementedError("historic hooks are not supported by AsyncPluginManager")

AsyncPipelineCaller dataclass

Bases: AsyncHookCaller

A pipeline async hook's typed caller: await a call to get R.

Source code in src/pluginkit/aio.py 
136
137
138
139
140
141
class AsyncPipelineCaller[**P, R](AsyncHookCaller):
    """A pipeline async hook's typed caller: `await` a call to get `R`."""

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Await the pipeline hook, returning the threaded value `R`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller
Methods:
__call__(*args, **kwargs) async

Await the pipeline hook, returning the threaded value R.

Source code in src/pluginkit/aio.py 
139
140
141
async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Await the pipeline hook, returning the threaded value `R`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller

AsyncPluginManager

Bases: PluginManager

A PluginManager whose hooks are awaited; impls may be coroutine functions.

Source code in src/pluginkit/aio.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class AsyncPluginManager(PluginManager):
    """A PluginManager whose hooks are awaited; impls may be coroutine functions."""

    def _make_caller(
        self, name: str, spec: ExtensionPointOpts, params: tuple[str, ...], defaults: dict[str, Any]
    ) -> HookCaller:
        """Build an AsyncHookCaller instead of the synchronous one."""
        return AsyncHookCaller(name=name, spec=spec, params=params, defaults=defaults)

    @overload  # type: ignore[override]  # async manager returns awaitable callers
    def caller[**P, R](self, spec: FirstResultSpec[P, R]) -> AsyncFirstResultCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: PipelineSpec[P, R]) -> AsyncPipelineCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: CollectingSpec[P, R]) -> AsyncCollectingCaller[P, R]: ...
    def caller(  # pyright: ignore[reportIncompatibleMethodOverride]  # async returns awaitable callers
        self, spec: object
    ) -> HookCaller:
        """Return the typed async caller for an `@extension_point`-decorated function."""
        return self._caller(spec)
Methods:
caller(spec) 
caller(
    spec: FirstResultSpec[P, R],
) -> AsyncFirstResultCaller[P, R]

caller(
    spec: PipelineSpec[P, R],
) -> AsyncPipelineCaller[P, R]

caller(
    spec: CollectingSpec[P, R],
) -> AsyncCollectingCaller[P, R]

Return the typed async caller for an @extension_point-decorated function.

Source code in src/pluginkit/aio.py 
159
160
161
162
163
def caller(  # pyright: ignore[reportIncompatibleMethodOverride]  # async returns awaitable callers
    self, spec: object
) -> HookCaller:
    """Return the typed async caller for an `@extension_point`-decorated function."""
    return self._caller(spec)

PluginValidationError

Bases: Exception

Raised when a plugin or one of its hook implementations is invalid.

Source code in src/pluginkit/exceptions.py 
 4
 5
 6
 7
 8
 9
10
class PluginValidationError(Exception):
    """Raised when a plugin or one of its hook implementations is invalid."""

    def __init__(self, plugin_name: str, message: str) -> None:
        """Record the offending plugin name alongside the message."""
        self.plugin_name = plugin_name
        super().__init__(f"plugin {plugin_name!r}: {message}")
Methods:
__init__(plugin_name, message)

Record the offending plugin name alongside the message.

Source code in src/pluginkit/exceptions.py 
 7
 8
 9
10
def __init__(self, plugin_name: str, message: str) -> None:
    """Record the offending plugin name alongside the message."""
    self.plugin_name = plugin_name
    super().__init__(f"plugin {plugin_name!r}: {message}")

CollectingCaller dataclass

Bases: HookCaller

A collecting hook's typed caller: a call returns list[R].

Source code in src/pluginkit/manager.py 
323
324
325
326
327
328
class CollectingCaller[**P, R](HookCaller):
    """A collecting hook's typed caller: a call returns `list[R]`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
        """Call the collecting hook, returning each impl's result as `list[R]`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Call the collecting hook, returning each impl's result as list[R].

Source code in src/pluginkit/manager.py 
326
327
328
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
    """Call the collecting hook, returning each impl's result as `list[R]`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

FirstResultCaller dataclass

Bases: HookCaller

A firstresult hook's typed caller: a call returns R | None.

Source code in src/pluginkit/manager.py 
331
332
333
334
335
336
class FirstResultCaller[**P, R](HookCaller):
    """A firstresult hook's typed caller: a call returns `R | None`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
        """Call the firstresult hook, returning the first non-None `R` or `None`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Call the firstresult hook, returning the first non-None R or None.

Source code in src/pluginkit/manager.py 
334
335
336
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
    """Call the firstresult hook, returning the first non-None `R` or `None`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

HistoricCaller dataclass

Bases: HookCaller

A historic hook's typed caller. Replay it with call_historic({...}).

Calling it directly raises - historic hooks have no plain call form - so the typed __call__ is NoReturn rather than a value it never produces.

Source code in src/pluginkit/manager.py 
347
348
349
350
351
352
353
354
355
356
class HistoricCaller[**P, R](HookCaller):
    """A historic hook's typed caller. Replay it with `call_historic({...})`.

    Calling it directly raises - historic hooks have no plain call form - so the
    typed `__call__` is `NoReturn` rather than a value it never produces.
    """

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> NoReturn:
        """Historic hooks cannot be called directly; use `call_historic`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Historic hooks cannot be called directly; use call_historic.

Source code in src/pluginkit/manager.py 
354
355
356
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> NoReturn:
    """Historic hooks cannot be called directly; use `call_historic`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

HookCaller dataclass

Holds every implementation of one hook and dispatches calls to them.

Source code in src/pluginkit/manager.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@dataclass(slots=True)
class HookCaller:
    """Holds every implementation of one hook and dispatches calls to them."""

    name: str
    spec: ExtensionPointOpts
    params: tuple[str, ...] = ()
    argnames: frozenset[str] = frozenset()
    # Default values for spec params that declare one. A call may omit these (the
    # branded caller's ParamSpec makes them optional); they are filled in at call
    # time so the type checker and the runtime agree.
    defaults: dict[str, Any] = field(default_factory=dict)
    _impls: list[HookImpl] = field(default_factory=list)
    _wrappers: list[HookImpl] = field(default_factory=list)
    _nonwrappers: list[HookImpl] = field(default_factory=list)
    _history: list[tuple[dict[str, Any], Callable[[Any], None] | None]] = field(default_factory=list)

    def __post_init__(self) -> None:
        """Derive the argument-name set from the ordered parameters when given."""
        if self.params and not self.argnames:
            self.argnames = frozenset(self.params)

    def check_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
        """Fill any omitted defaulted args, then validate the call against the spec.

        Returns the completed kwargs (defaults filled). Spec params with a default
        are optional at the call site; required params and unknown args are still
        rejected.
        """
        if self.defaults:
            kwargs = {**self.defaults, **kwargs}
        # dict_keys compares as a set against the frozenset without allocating one.
        if kwargs.keys() == self.argnames:
            return kwargs
        provided = frozenset(kwargs)
        problems: list[str] = []
        missing = self.argnames - provided
        unknown = provided - self.argnames
        if missing:
            problems.append(f"missing {sorted(missing)}")
        if unknown:
            problems.append(f"unknown {sorted(unknown)}")
        raise TypeError(f"hook {self.name!r} called with {'; '.join(problems)}; expects {sorted(self.argnames)}")

    def add_impl(self, impl: HookImpl) -> None:
        """Add an impl in priority order and replay any historic calls to it."""
        impl.passthrough = impl.accepts == self.argnames
        self._impls.append(impl)
        self._reindex()
        for kwargs, callback in self._history:
            outcome = impl.call(kwargs)
            if outcome is not None and callback is not None:
                callback(outcome)

    def remove_plugin(self, plugin_name: str) -> bool:
        """Drop every impl contributed by a plugin; return True if any were removed."""
        before = len(self._impls)
        self._impls = [impl for impl in self._impls if impl.plugin_name != plugin_name]
        removed = len(self._impls) != before
        if removed:
            self._reindex()
        return removed

    def has_plugin(self, plugin_name: str) -> bool:
        """Return whether the named plugin contributes any impl to this hook."""
        return any(impl.plugin_name == plugin_name for impl in self._impls)

    def _prepare_extra(self, functions: list[Callable[..., Any]]) -> list[HookImpl]:
        """Build one-off impls for call_extra, validating their args against the spec."""
        extra: list[HookImpl] = []
        for function in functions:
            impl = HookImpl.from_function("<call_extra>", function, ExtensionOpts())
            unknown = impl.accepts - self.argnames
            if unknown:
                raise TypeError(f"call_extra impl for {self.name!r} declares unknown argument(s) {sorted(unknown)}")
            impl.passthrough = impl.accepts == self.argnames
            extra.append(impl)
        return extra

    def _bind(self, args: tuple[Any, ...], kwargs: dict[str, Any]) -> dict[str, Any]:
        """Bind positional args to the spec's params (in order) and merge with kwargs.

        Lets a typed caller be invoked positionally - `caller(value)` as well as
        `caller(name=value)` - matching what the ParamSpec advertises.
        """
        if not args:
            return kwargs
        if len(args) > len(self.params):
            raise TypeError(f"hook {self.name!r} takes at most {len(self.params)} positional argument(s)")
        positional = dict(zip(self.params, args, strict=False))
        clash = positional.keys() & kwargs.keys()
        if clash:
            raise TypeError(f"hook {self.name!r} got multiple values for {sorted(clash)}")
        return {**positional, **kwargs}

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        """Call the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return self._execute(kwargs, self._nonwrappers)

    def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
        """Call the hook with extra one-off implementations that are not registered.

        The extra functions run as normal-priority implementations for this call
        only, ordered after the already-registered ones. Useful for tests and for
        injecting a temporary implementation without mutating the manager.
        """
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(kwargs)
        combined = sorted(
            [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
        )
        return self._execute(kwargs, combined)

    def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
        """Call a historic hook now and remember it for plugins registered later."""
        if not self.spec.historic:
            raise TypeError(f"hook {self.name!r} is not historic")
        kwargs = self.check_arguments(kwargs)
        self._history.append((kwargs, result_callback))
        for outcome in self._collect(kwargs):
            if result_callback is not None:
                result_callback(outcome)

    def _reindex(self) -> None:
        """Re-sort impls by priority and refresh the wrapper / non-wrapper split."""
        # Stable sort keeps registration order within each priority bucket.
        self._impls.sort(key=lambda candidate: candidate.order_key)
        self._wrappers = [impl for impl in self._impls if impl.opts.wrapper]
        self._nonwrappers = [impl for impl in self._impls if not impl.opts.wrapper]

    def _execute(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Run the inner impls, wrapped by any wrappers, unwinding exception-safely."""
        # Fast path: with no wrappers there is nothing to unwind, so skip the
        # try/except and generator bookkeeping entirely and let errors propagate.
        if not self._wrappers:
            return self._core(kwargs, nonwrappers)
        started: list[Generator[Any, Any, Any]] = []
        try:
            for wrapper in self._wrappers:
                generator = wrapper.call(kwargs)
                if not isinstance(generator, GeneratorType):
                    raise TypeError(f"wrapper {wrapper.plugin_name}.{self.name} must be a generator function")
                next(generator)  # advance to the yield
                started.append(generator)
            result = self._core(kwargs, nonwrappers)
        except BaseException as exc:  # noqa: BLE001 - re-raised after wrappers observe it
            return self._teardown(started, exc=exc)
        return self._teardown(started, result=result)

    def _core(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Apply the spec's dispatch strategy to the non-wrapper impls."""
        if self.spec.pipeline:
            return self._run_pipeline(kwargs, nonwrappers)
        if self.spec.firstresult:
            for impl in nonwrappers:
                outcome = impl.call(kwargs)
                if outcome is not None:
                    return outcome
            return None
        results: list[Any] = []
        for impl in nonwrappers:
            outcome = impl.call(kwargs)
            if outcome is not None:
                results.append(outcome)
        return results

    def _run_pipeline(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Thread the first argument through each impl, feeding its result to the next."""
        param = self.params[0]
        value = kwargs[param]
        current = dict(kwargs)
        for impl in nonwrappers:
            current[param] = value
            outcome = impl.call(current)
            if outcome is not None:  # None means "pass the value through unchanged"
                value = outcome
        return value

    def _collect(self, kwargs: dict[str, Any]) -> list[Any]:
        """Return the non-None results of the non-wrapper impls as a list."""
        return list(self._collect_iter(kwargs, self._nonwrappers))

    def _collect_iter(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Iterator[Any]:
        """Yield non-None results from the given non-wrapper impls, honouring firstresult."""
        for impl in nonwrappers:
            outcome = impl.call(kwargs)
            if outcome is None:
                continue
            yield outcome
            if self.spec.firstresult:
                return

    def _teardown(
        self, started: list[Generator[Any, Any, Any]], *, result: Any = _UNSET, exc: BaseException | None = None
    ) -> Any:
        """Resume each wrapper in reverse, letting it replace the result or handle the error."""
        for generator in reversed(started):
            try:
                if exc is not None:
                    generator.throw(exc)
                else:
                    generator.send(result)
            except StopIteration as stop:
                # A wrapper that returns after the yield ends here.
                if exc is not None:
                    # The wrapper swallowed the exception and supplied a result.
                    exc = None
                    result = stop.value
                elif stop.value is not None:
                    result = stop.value
            except BaseException as new_exc:  # noqa: BLE001 - propagate the wrapper's error onward
                exc = new_exc
            else:
                # The generator yielded a second time, violating the one-yield contract.
                # Capture the error but keep unwinding so the remaining wrappers still
                # tear down; the error propagates through them and is raised at the end.
                generator.close()
                exc = RuntimeError(f"wrapper for {self.name!r} must yield exactly once")
        if exc is not None:
            raise exc
        return result

    def implementations(self) -> list[HookImpl]:
        """Return this hook's implementations in call order (wrappers excluded)."""
        return list(self._nonwrappers)

    def __repr__(self) -> str:
        """Show the hook name and how many implementations it has."""
        return f"<HookCaller {self.name!r} impls={len(self._impls)}>"
Methods:
__post_init__()

Derive the argument-name set from the ordered parameters when given.

Source code in src/pluginkit/manager.py 
101
102
103
104
def __post_init__(self) -> None:
    """Derive the argument-name set from the ordered parameters when given."""
    if self.params and not self.argnames:
        self.argnames = frozenset(self.params)
check_arguments(kwargs)

Fill any omitted defaulted args, then validate the call against the spec.

Returns the completed kwargs (defaults filled). Spec params with a default are optional at the call site; required params and unknown args are still rejected.

Source code in src/pluginkit/manager.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def check_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
    """Fill any omitted defaulted args, then validate the call against the spec.

    Returns the completed kwargs (defaults filled). Spec params with a default
    are optional at the call site; required params and unknown args are still
    rejected.
    """
    if self.defaults:
        kwargs = {**self.defaults, **kwargs}
    # dict_keys compares as a set against the frozenset without allocating one.
    if kwargs.keys() == self.argnames:
        return kwargs
    provided = frozenset(kwargs)
    problems: list[str] = []
    missing = self.argnames - provided
    unknown = provided - self.argnames
    if missing:
        problems.append(f"missing {sorted(missing)}")
    if unknown:
        problems.append(f"unknown {sorted(unknown)}")
    raise TypeError(f"hook {self.name!r} called with {'; '.join(problems)}; expects {sorted(self.argnames)}")
add_impl(impl)

Add an impl in priority order and replay any historic calls to it.

Source code in src/pluginkit/manager.py 
128
129
130
131
132
133
134
135
136
def add_impl(self, impl: HookImpl) -> None:
    """Add an impl in priority order and replay any historic calls to it."""
    impl.passthrough = impl.accepts == self.argnames
    self._impls.append(impl)
    self._reindex()
    for kwargs, callback in self._history:
        outcome = impl.call(kwargs)
        if outcome is not None and callback is not None:
            callback(outcome)
remove_plugin(plugin_name)

Drop every impl contributed by a plugin; return True if any were removed.

Source code in src/pluginkit/manager.py 
138
139
140
141
142
143
144
145
def remove_plugin(self, plugin_name: str) -> bool:
    """Drop every impl contributed by a plugin; return True if any were removed."""
    before = len(self._impls)
    self._impls = [impl for impl in self._impls if impl.plugin_name != plugin_name]
    removed = len(self._impls) != before
    if removed:
        self._reindex()
    return removed
has_plugin(plugin_name)

Return whether the named plugin contributes any impl to this hook.

Source code in src/pluginkit/manager.py 
147
148
149
def has_plugin(self, plugin_name: str) -> bool:
    """Return whether the named plugin contributes any impl to this hook."""
    return any(impl.plugin_name == plugin_name for impl in self._impls)
__call__(*args, **kwargs)

Call the hook: a list, a single value (firstresult), or the threaded value (pipeline).

Source code in src/pluginkit/manager.py 
179
180
181
182
183
184
def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Call the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return self._execute(kwargs, self._nonwrappers)
call_extra(functions, kwargs)

Call the hook with extra one-off implementations that are not registered.

The extra functions run as normal-priority implementations for this call only, ordered after the already-registered ones. Useful for tests and for injecting a temporary implementation without mutating the manager.

Source code in src/pluginkit/manager.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
    """Call the hook with extra one-off implementations that are not registered.

    The extra functions run as normal-priority implementations for this call
    only, ordered after the already-registered ones. Useful for tests and for
    injecting a temporary implementation without mutating the manager.
    """
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(kwargs)
    combined = sorted(
        [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
    )
    return self._execute(kwargs, combined)
call_historic(kwargs, result_callback=None)

Call a historic hook now and remember it for plugins registered later.

Source code in src/pluginkit/manager.py 
201
202
203
204
205
206
207
208
209
def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
    """Call a historic hook now and remember it for plugins registered later."""
    if not self.spec.historic:
        raise TypeError(f"hook {self.name!r} is not historic")
    kwargs = self.check_arguments(kwargs)
    self._history.append((kwargs, result_callback))
    for outcome in self._collect(kwargs):
        if result_callback is not None:
            result_callback(outcome)
implementations()

Return this hook's implementations in call order (wrappers excluded).

Source code in src/pluginkit/manager.py 
310
311
312
def implementations(self) -> list[HookImpl]:
    """Return this hook's implementations in call order (wrappers excluded)."""
    return list(self._nonwrappers)
__repr__()

Show the hook name and how many implementations it has.

Source code in src/pluginkit/manager.py 
314
315
316
def __repr__(self) -> str:
    """Show the hook name and how many implementations it has."""
    return f"<HookCaller {self.name!r} impls={len(self._impls)}>"

HookImpl dataclass

One plugin's implementation of a hook, plus the kwargs it accepts.

Source code in src/pluginkit/manager.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(slots=True)
class HookImpl:
    """One plugin's implementation of a hook, plus the kwargs it accepts."""

    plugin_name: str
    function: Callable[..., Any]
    opts: ExtensionOpts
    accepts: frozenset[str]
    params: tuple[str, ...]
    # Set by the caller once it knows the hook's full argument set: True when this
    # impl declares exactly those arguments, so kwargs can be forwarded directly.
    passthrough: bool = False

    @classmethod
    def from_function(cls, plugin_name: str, function: Callable[..., Any], opts: ExtensionOpts) -> Self:
        """Build an impl, recording which keyword arguments the function declares."""
        params = tuple(inspect.signature(function).parameters)
        return cls(plugin_name=plugin_name, function=function, opts=opts, accepts=frozenset(params), params=params)

    def call(self, kwargs: dict[str, Any]) -> Any:
        """Invoke the function, passing only the arguments it declares.

        The caller guarantees every declared argument is present in kwargs, so the
        common "takes all the spec's arguments" case forwards kwargs directly and a
        subset impl indexes the few it wants - both avoiding a membership scan.
        """
        if self.passthrough:
            return self.function(**kwargs)
        return self.function(**{name: kwargs[name] for name in self.params})

    @property
    def order_key(self) -> int:
        """Sort key: tryfirst impls run first (0), normal next (1), trylast last (2)."""
        match self.opts:
            case ExtensionOpts(tryfirst=True):
                return 0
            case ExtensionOpts(trylast=True):
                return 2
            case _:
                return 1
Attributes
order_key property

Sort key: tryfirst impls run first (0), normal next (1), trylast last (2).

Methods:
from_function(plugin_name, function, opts) classmethod

Build an impl, recording which keyword arguments the function declares.

Source code in src/pluginkit/manager.py 
55
56
57
58
59
@classmethod
def from_function(cls, plugin_name: str, function: Callable[..., Any], opts: ExtensionOpts) -> Self:
    """Build an impl, recording which keyword arguments the function declares."""
    params = tuple(inspect.signature(function).parameters)
    return cls(plugin_name=plugin_name, function=function, opts=opts, accepts=frozenset(params), params=params)
call(kwargs)

Invoke the function, passing only the arguments it declares.

The caller guarantees every declared argument is present in kwargs, so the common "takes all the spec's arguments" case forwards kwargs directly and a subset impl indexes the few it wants - both avoiding a membership scan.

Source code in src/pluginkit/manager.py 
61
62
63
64
65
66
67
68
69
70
def call(self, kwargs: dict[str, Any]) -> Any:
    """Invoke the function, passing only the arguments it declares.

    The caller guarantees every declared argument is present in kwargs, so the
    common "takes all the spec's arguments" case forwards kwargs directly and a
    subset impl indexes the few it wants - both avoiding a membership scan.
    """
    if self.passthrough:
        return self.function(**kwargs)
    return self.function(**{name: kwargs[name] for name in self.params})

HookRelay

Attribute-style access to hook callers, e.g. pm.hook.add_ingredients(...).

Source code in src/pluginkit/manager.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
class HookRelay:
    """Attribute-style access to hook callers, e.g. pm.hook.add_ingredients(...)."""

    def __init__(self) -> None:
        """Start with no registered callers."""
        self._callers: dict[str, HookCaller] = {}

    def _add_caller(self, caller: HookCaller) -> None:
        """Register a caller under its hook name."""
        self._callers[caller.name] = caller

    def _get_caller(self, name: str) -> HookCaller | None:
        """Return the caller for a hook name, or None if undefined."""
        return self._callers.get(name)

    def _all_callers(self) -> list[HookCaller]:
        """Return every registered caller."""
        return list(self._callers.values())

    def __getattr__(self, name: str) -> HookCaller:
        """Resolve pm.hook.<name> to its HookCaller, or raise AttributeError."""
        try:
            return self._callers[name]
        except KeyError:
            raise AttributeError(f"no hook named {name!r}") from None
Methods:
__init__()

Start with no registered callers.

Source code in src/pluginkit/manager.py 
362
363
364
def __init__(self) -> None:
    """Start with no registered callers."""
    self._callers: dict[str, HookCaller] = {}
__getattr__(name)

Resolve pm.hook. to its HookCaller, or raise AttributeError.

Source code in src/pluginkit/manager.py 
378
379
380
381
382
383
def __getattr__(self, name: str) -> HookCaller:
    """Resolve pm.hook.<name> to its HookCaller, or raise AttributeError."""
    try:
        return self._callers[name]
    except KeyError:
        raise AttributeError(f"no hook named {name!r}") from None

PipelineCaller dataclass

Bases: HookCaller

A pipeline hook's typed caller: a call returns R.

Source code in src/pluginkit/manager.py 
339
340
341
342
343
344
class PipelineCaller[**P, R](HookCaller):
    """A pipeline hook's typed caller: a call returns `R`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Call the pipeline hook, returning the threaded value `R`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Call the pipeline hook, returning the threaded value R.

Source code in src/pluginkit/manager.py 
342
343
344
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Call the pipeline hook, returning the threaded value `R`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

PluginManager

Registers plugins and exposes their hooks via a HookRelay.

Source code in src/pluginkit/manager.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
class PluginManager:
    """Registers plugins and exposes their hooks via a HookRelay."""

    def __init__(self, project_name: str) -> None:
        """Bind the manager to a project name shared with the markers."""
        self.project_name = project_name
        self.hook = HookRelay()
        self._spec_attribute = f"{project_name}_extension_point"
        self._impl_attribute = f"{project_name}_extension"
        self._name2plugin: dict[str, object] = {}
        self._blocked: set[str] = set()
        self._lock = threading.RLock()

    def add_extension_points(self, namespace: object) -> None:
        """Scan a module (or object) for extension points and create callers."""
        with self._lock:
            for member_name in dir(namespace):
                member = getattr(namespace, member_name)
                spec = getattr(member, self._spec_attribute, None)
                if not isinstance(spec, ExtensionPointOpts):
                    continue
                signature = inspect.signature(member)
                params = tuple(signature.parameters)
                defaults = {
                    name: parameter.default
                    for name, parameter in signature.parameters.items()
                    if parameter.default is not inspect.Parameter.empty
                }
                self._validate_spec(member_name, spec, params)
                self.hook._add_caller(self._make_caller(member_name, spec, params, defaults))

    def _make_caller(
        self, name: str, spec: ExtensionPointOpts, params: tuple[str, ...], defaults: dict[str, Any]
    ) -> HookCaller:
        """Build the caller for a spec; overridden by AsyncPluginManager."""
        return HookCaller(name=name, spec=spec, params=params, defaults=defaults)

    @overload
    def caller[**P, R](self, spec: FirstResultSpec[P, R]) -> FirstResultCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: PipelineSpec[P, R]) -> PipelineCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: HistoricSpec[P, R]) -> HistoricCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: CollectingSpec[P, R]) -> CollectingCaller[P, R]: ...
    def caller(self, spec: object) -> HookCaller:
        """Return the typed caller for an `@extension_point`-decorated function.

        The result is a plain `HookCaller`, but its static type carries the extension
        point's dispatch mode, so a call returns `list[R]` (collecting), `R | None`
        (firstresult), or `R` (pipeline) - derived from the declaration, not asserted.
        """
        return self._caller(spec)

    def _caller(self, spec: object) -> HookCaller:
        """Resolve an extension point to its registered caller (shared by subclasses)."""
        name = getattr(spec, "__name__", None)
        if not isinstance(name, str):
            raise TypeError("caller() expects an @extension_point-decorated function")
        found = self.hook._get_caller(name)
        if found is None:
            raise PluginValidationError(
                self.project_name, f"unknown extension point {name!r}; call add_extension_points() first"
            )
        return found

    @staticmethod
    def _validate_spec(name: str, spec: ExtensionPointOpts, params: tuple[str, ...]) -> None:
        """Reject contradictory or impossible spec option combinations."""
        modes = [
            mode
            for mode, on in (
                ("firstresult", spec.firstresult),
                ("historic", spec.historic),
                ("pipeline", spec.pipeline),
            )
            if on
        ]
        if len(modes) > 1:
            raise ValueError(f"hook {name!r} cannot combine {' and '.join(modes)}")
        if spec.pipeline and not params:
            raise ValueError(f"pipeline hook {name!r} must declare at least one argument to thread through")

    def register(self, plugin: object, name: str | None = None) -> str:
        """Register a plugin object, wiring up every hook implementation it carries."""
        with self._lock:
            plugin_name = name or self.get_canonical_name(plugin)
            if plugin_name in self._blocked:
                raise ValueError(f"plugin {plugin_name!r} is blocked")
            if plugin_name in self._name2plugin:
                raise ValueError(f"plugin name {plugin_name!r} is already registered")
            if any(existing is plugin for existing in self._name2plugin.values()):
                raise ValueError(f"plugin object {plugin!r} is already registered")

            impls = self._collect_impls(plugin_name, plugin)
            self._name2plugin[plugin_name] = plugin
            try:
                for caller, impl in impls:
                    caller.add_impl(impl)
            except BaseException:
                # add_impl can fail mid-loop (e.g. a historic replay raising). Roll the
                # partial wiring back so registration is all-or-nothing.
                self._name2plugin.pop(plugin_name, None)
                for caller in self.hook._all_callers():
                    caller.remove_plugin(plugin_name)
                raise
            return plugin_name

    def unregister(self, name_or_plugin: str | object) -> object | None:
        """Remove a plugin by name or by object; return the removed plugin or None."""
        with self._lock:
            name = name_or_plugin if isinstance(name_or_plugin, str) else self.get_name(name_or_plugin)
            if name is None:
                return None
            plugin = self._name2plugin.pop(name, None)
            if plugin is None:
                return None
            for caller in self.hook._all_callers():
                caller.remove_plugin(name)
            return plugin

    def set_blocked(self, name: str) -> None:
        """Block a plugin name: unregister it if present and refuse future registration."""
        with self._lock:
            self._blocked.add(name)
            self.unregister(name)

    def is_blocked(self, name: str) -> bool:
        """Return whether a plugin name is blocked."""
        return name in self._blocked

    def is_registered(self, plugin: object) -> bool:
        """Return whether a plugin object is currently registered."""
        return any(existing is plugin for existing in self._name2plugin.values())

    def get_plugin(self, name: str) -> object | None:
        """Return the plugin registered under a name, or None."""
        return self._name2plugin.get(name)

    def get_name(self, plugin: object) -> str | None:
        """Return the registered name of a plugin object, or None."""
        for registered_name, registered_plugin in self._name2plugin.items():
            if registered_plugin is plugin:
                return registered_name
        return None

    def get_canonical_name(self, plugin: object) -> str:
        """Derive a default name for a plugin from its __name__ or type."""
        return getattr(plugin, "__name__", None) or type(plugin).__name__

    def plugin_names(self) -> list[str]:
        """Return the names of all registered plugins, in registration order."""
        return list(self._name2plugin)

    def get_hookcallers(self, plugin: object) -> list[HookCaller] | None:
        """Return the hooks a registered plugin contributes to, or None if unknown."""
        name = self.get_name(plugin)
        if name is None:
            return None
        return [caller for caller in self.hook._all_callers() if caller.has_plugin(name)]

    def __repr__(self) -> str:
        """Show the project name and number of registered plugins."""
        return f"<PluginManager {self.project_name!r} plugins={len(self._name2plugin)}>"

    def load_entrypoints(self, group: str, *, ignore_errors: bool = False) -> int:
        """Discover and register external plugins advertised under an entry-point group.

        Args:
            group: The entry-point group name to scan.
            ignore_errors: When True, skip plugins that fail to load or register
                instead of raising, so one broken plugin cannot block discovery.

        Returns:
            The number of plugins successfully registered.

        Note:
            With ``ignore_errors=False``, a failure part-way through leaves the
            plugins registered before it registered; this method does not roll back
            across plugins.
        """
        count = 0
        for entry_point in entry_points(group=group):
            if entry_point.name in self._name2plugin or entry_point.name in self._blocked:
                continue
            try:
                plugin = entry_point.load()
                self.register(plugin, name=entry_point.name)
            except Exception as error:
                if ignore_errors:
                    continue
                raise PluginValidationError(entry_point.name, f"failed to load entry point: {error}") from error
            count += 1
        return count

    def _collect_impls(self, plugin_name: str, plugin: object) -> list[tuple[HookCaller, HookImpl]]:
        """Find and validate every hook implementation a plugin carries."""
        collected: list[tuple[HookCaller, HookImpl]] = []
        for member_name in dir(plugin):
            member = getattr(plugin, member_name)
            opts = getattr(member, self._impl_attribute, None)
            if not isinstance(opts, ExtensionOpts):
                continue
            hook_name = opts.target or member_name
            caller = self.hook._get_caller(hook_name)
            if caller is None:
                if opts.optional:
                    continue
                raise PluginValidationError(plugin_name, f"implements unknown extension point {hook_name!r}")
            if opts.wrapper and caller.spec.historic:
                raise PluginValidationError(plugin_name, f"historic hook {hook_name!r} cannot have a wrapper")
            impl = HookImpl.from_function(plugin_name, member, opts)
            unknown = impl.accepts - caller.argnames
            if unknown:
                raise PluginValidationError(
                    plugin_name,
                    f"hook {hook_name!r} impl declares unknown argument(s) {sorted(unknown)}; "
                    f"spec accepts {sorted(caller.argnames)}",
                )
            collected.append((caller, impl))
        return collected
Methods:
__init__(project_name)

Bind the manager to a project name shared with the markers.

Source code in src/pluginkit/manager.py 
389
390
391
392
393
394
395
396
397
def __init__(self, project_name: str) -> None:
    """Bind the manager to a project name shared with the markers."""
    self.project_name = project_name
    self.hook = HookRelay()
    self._spec_attribute = f"{project_name}_extension_point"
    self._impl_attribute = f"{project_name}_extension"
    self._name2plugin: dict[str, object] = {}
    self._blocked: set[str] = set()
    self._lock = threading.RLock()
add_extension_points(namespace)

Scan a module (or object) for extension points and create callers.

Source code in src/pluginkit/manager.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def add_extension_points(self, namespace: object) -> None:
    """Scan a module (or object) for extension points and create callers."""
    with self._lock:
        for member_name in dir(namespace):
            member = getattr(namespace, member_name)
            spec = getattr(member, self._spec_attribute, None)
            if not isinstance(spec, ExtensionPointOpts):
                continue
            signature = inspect.signature(member)
            params = tuple(signature.parameters)
            defaults = {
                name: parameter.default
                for name, parameter in signature.parameters.items()
                if parameter.default is not inspect.Parameter.empty
            }
            self._validate_spec(member_name, spec, params)
            self.hook._add_caller(self._make_caller(member_name, spec, params, defaults))
caller(spec) 
caller(
    spec: FirstResultSpec[P, R],
) -> FirstResultCaller[P, R]

caller(spec: PipelineSpec[P, R]) -> PipelineCaller[P, R]

caller(spec: HistoricSpec[P, R]) -> HistoricCaller[P, R]

caller(
    spec: CollectingSpec[P, R],
) -> CollectingCaller[P, R]

Return the typed caller for an @extension_point-decorated function.

The result is a plain HookCaller, but its static type carries the extension point's dispatch mode, so a call returns list[R] (collecting), R | None (firstresult), or R (pipeline) - derived from the declaration, not asserted.

Source code in src/pluginkit/manager.py 
431
432
433
434
435
436
437
438
def caller(self, spec: object) -> HookCaller:
    """Return the typed caller for an `@extension_point`-decorated function.

    The result is a plain `HookCaller`, but its static type carries the extension
    point's dispatch mode, so a call returns `list[R]` (collecting), `R | None`
    (firstresult), or `R` (pipeline) - derived from the declaration, not asserted.
    """
    return self._caller(spec)
register(plugin, name=None)

Register a plugin object, wiring up every hook implementation it carries.

Source code in src/pluginkit/manager.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def register(self, plugin: object, name: str | None = None) -> str:
    """Register a plugin object, wiring up every hook implementation it carries."""
    with self._lock:
        plugin_name = name or self.get_canonical_name(plugin)
        if plugin_name in self._blocked:
            raise ValueError(f"plugin {plugin_name!r} is blocked")
        if plugin_name in self._name2plugin:
            raise ValueError(f"plugin name {plugin_name!r} is already registered")
        if any(existing is plugin for existing in self._name2plugin.values()):
            raise ValueError(f"plugin object {plugin!r} is already registered")

        impls = self._collect_impls(plugin_name, plugin)
        self._name2plugin[plugin_name] = plugin
        try:
            for caller, impl in impls:
                caller.add_impl(impl)
        except BaseException:
            # add_impl can fail mid-loop (e.g. a historic replay raising). Roll the
            # partial wiring back so registration is all-or-nothing.
            self._name2plugin.pop(plugin_name, None)
            for caller in self.hook._all_callers():
                caller.remove_plugin(plugin_name)
            raise
        return plugin_name
unregister(name_or_plugin)

Remove a plugin by name or by object; return the removed plugin or None.

Source code in src/pluginkit/manager.py 
494
495
496
497
498
499
500
501
502
503
504
505
def unregister(self, name_or_plugin: str | object) -> object | None:
    """Remove a plugin by name or by object; return the removed plugin or None."""
    with self._lock:
        name = name_or_plugin if isinstance(name_or_plugin, str) else self.get_name(name_or_plugin)
        if name is None:
            return None
        plugin = self._name2plugin.pop(name, None)
        if plugin is None:
            return None
        for caller in self.hook._all_callers():
            caller.remove_plugin(name)
        return plugin
set_blocked(name)

Block a plugin name: unregister it if present and refuse future registration.

Source code in src/pluginkit/manager.py 
507
508
509
510
511
def set_blocked(self, name: str) -> None:
    """Block a plugin name: unregister it if present and refuse future registration."""
    with self._lock:
        self._blocked.add(name)
        self.unregister(name)
is_blocked(name)

Return whether a plugin name is blocked.

Source code in src/pluginkit/manager.py 
513
514
515
def is_blocked(self, name: str) -> bool:
    """Return whether a plugin name is blocked."""
    return name in self._blocked
is_registered(plugin)

Return whether a plugin object is currently registered.

Source code in src/pluginkit/manager.py 
517
518
519
def is_registered(self, plugin: object) -> bool:
    """Return whether a plugin object is currently registered."""
    return any(existing is plugin for existing in self._name2plugin.values())
get_plugin(name)

Return the plugin registered under a name, or None.

Source code in src/pluginkit/manager.py 
521
522
523
def get_plugin(self, name: str) -> object | None:
    """Return the plugin registered under a name, or None."""
    return self._name2plugin.get(name)
get_name(plugin)

Return the registered name of a plugin object, or None.

Source code in src/pluginkit/manager.py 
525
526
527
528
529
530
def get_name(self, plugin: object) -> str | None:
    """Return the registered name of a plugin object, or None."""
    for registered_name, registered_plugin in self._name2plugin.items():
        if registered_plugin is plugin:
            return registered_name
    return None
get_canonical_name(plugin)

Derive a default name for a plugin from its name or type.

Source code in src/pluginkit/manager.py 
532
533
534
def get_canonical_name(self, plugin: object) -> str:
    """Derive a default name for a plugin from its __name__ or type."""
    return getattr(plugin, "__name__", None) or type(plugin).__name__
plugin_names()

Return the names of all registered plugins, in registration order.

Source code in src/pluginkit/manager.py 
536
537
538
def plugin_names(self) -> list[str]:
    """Return the names of all registered plugins, in registration order."""
    return list(self._name2plugin)
get_hookcallers(plugin)

Return the hooks a registered plugin contributes to, or None if unknown.

Source code in src/pluginkit/manager.py 
540
541
542
543
544
545
def get_hookcallers(self, plugin: object) -> list[HookCaller] | None:
    """Return the hooks a registered plugin contributes to, or None if unknown."""
    name = self.get_name(plugin)
    if name is None:
        return None
    return [caller for caller in self.hook._all_callers() if caller.has_plugin(name)]
__repr__()

Show the project name and number of registered plugins.

Source code in src/pluginkit/manager.py 
547
548
549
def __repr__(self) -> str:
    """Show the project name and number of registered plugins."""
    return f"<PluginManager {self.project_name!r} plugins={len(self._name2plugin)}>"
load_entrypoints(group, *, ignore_errors=False)

Discover and register external plugins advertised under an entry-point group.

Parameters:

Name Type Description Default
group str

The entry-point group name to scan.

 required
ignore_errors bool

When True, skip plugins that fail to load or register instead of raising, so one broken plugin cannot block discovery.

 False

Returns:

Type Description
int

The number of plugins successfully registered.
Note

With ignore_errors=False, a failure part-way through leaves the plugins registered before it registered; this method does not roll back across plugins.

Source code in src/pluginkit/manager.py 
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
def load_entrypoints(self, group: str, *, ignore_errors: bool = False) -> int:
    """Discover and register external plugins advertised under an entry-point group.

    Args:
        group: The entry-point group name to scan.
        ignore_errors: When True, skip plugins that fail to load or register
            instead of raising, so one broken plugin cannot block discovery.

    Returns:
        The number of plugins successfully registered.

    Note:
        With ``ignore_errors=False``, a failure part-way through leaves the
        plugins registered before it registered; this method does not roll back
        across plugins.
    """
    count = 0
    for entry_point in entry_points(group=group):
        if entry_point.name in self._name2plugin or entry_point.name in self._blocked:
            continue
        try:
            plugin = entry_point.load()
            self.register(plugin, name=entry_point.name)
        except Exception as error:
            if ignore_errors:
                continue
            raise PluginValidationError(entry_point.name, f"failed to load entry point: {error}") from error
        count += 1
    return count

CollectingSpec

A collecting extension point: a call collects each extension's R into list[R].

Source code in src/pluginkit/markers.py 
41
42
43
44
45
46
class CollectingSpec[**P, R]:
    """A collecting extension point: a call collects each extension's `R` into `list[R]`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
44
45
46
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

Extension

Creates the @extension decorator bound to a project name.

Source code in src/pluginkit/markers.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class Extension:
    """Creates the @extension decorator bound to a project name."""

    def __init__(self, project_name: str) -> None:
        """Bind the marker to a project name used for the stamped attribute."""
        self.project_name = project_name
        self.attribute = f"{project_name}_extension"

    @overload
    def __call__[F: Callable[..., Any]](self, function: F) -> F: ...
    @overload
    def __call__[F: Callable[..., Any]](
        self,
        function: None = ...,
        *,
        tryfirst: bool = ...,
        trylast: bool = ...,
        wrapper: bool = ...,
        optional: bool = ...,
        target: str | None = ...,
    ) -> Callable[[F], F]: ...
    def __call__[F: Callable[..., Any]](
        self,
        function: F | None = None,
        *,
        tryfirst: bool = False,
        trylast: bool = False,
        wrapper: bool = False,
        optional: bool = False,
        target: str | None = None,
    ) -> F | Callable[[F], F]:
        """Stamp ExtensionOpts onto the function; supports bare and called forms."""

        def mark(func: F) -> F:
            setattr(
                func,
                self.attribute,
                ExtensionOpts(
                    tryfirst=tryfirst,
                    trylast=trylast,
                    wrapper=wrapper,
                    optional=optional,
                    target=target,
                ),
            )
            return func

        return mark(function) if function is not None else mark
Methods:
__init__(project_name)

Bind the marker to a project name used for the stamped attribute.

Source code in src/pluginkit/markers.py 
121
122
123
124
def __init__(self, project_name: str) -> None:
    """Bind the marker to a project name used for the stamped attribute."""
    self.project_name = project_name
    self.attribute = f"{project_name}_extension"
__call__(function=None, *, tryfirst=False, trylast=False, wrapper=False, optional=False, target=None) 
__call__(function: F) -> F

__call__(
    function: None = ...,
    *,
    tryfirst: bool = ...,
    trylast: bool = ...,
    wrapper: bool = ...,
    optional: bool = ...,
    target: str | None = ...,
) -> Callable[[F], F]

Stamp ExtensionOpts onto the function; supports bare and called forms.

Source code in src/pluginkit/markers.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def __call__[F: Callable[..., Any]](
    self,
    function: F | None = None,
    *,
    tryfirst: bool = False,
    trylast: bool = False,
    wrapper: bool = False,
    optional: bool = False,
    target: str | None = None,
) -> F | Callable[[F], F]:
    """Stamp ExtensionOpts onto the function; supports bare and called forms."""

    def mark(func: F) -> F:
        setattr(
            func,
            self.attribute,
            ExtensionOpts(
                tryfirst=tryfirst,
                trylast=trylast,
                wrapper=wrapper,
                optional=optional,
                target=target,
            ),
        )
        return func

    return mark(function) if function is not None else mark

ExtensionOpts dataclass

Options attached to an extension.

Source code in src/pluginkit/markers.py 
30
31
32
33
34
35
36
37
38
@dataclass(frozen=True, slots=True)
class ExtensionOpts:
    """Options attached to an extension."""

    tryfirst: bool = False
    trylast: bool = False
    wrapper: bool = False
    optional: bool = False
    target: str | None = None

ExtensionPoint

Creates the @extension_point decorator bound to a project name.

Source code in src/pluginkit/markers.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class ExtensionPoint:
    """Creates the @extension_point decorator bound to a project name."""

    def __init__(self, project_name: str) -> None:
        """Bind the marker to a project name used for the stamped attribute."""
        self.project_name = project_name
        self.attribute = f"{project_name}_extension_point"

    @overload
    def __call__[**P, R](self, function: Callable[P, R]) -> CollectingSpec[P, R]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, firstresult: Literal[True], historic: bool = ...
    ) -> Callable[[Callable[P, R]], FirstResultSpec[P, R]]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, pipeline: Literal[True], historic: bool = ...
    ) -> Callable[[Callable[P, R]], PipelineSpec[P, R]]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, historic: Literal[True]
    ) -> Callable[[Callable[P, R]], HistoricSpec[P, R]]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, historic: Literal[False] = ...
    ) -> Callable[[Callable[P, R]], CollectingSpec[P, R]]: ...
    def __call__(
        self,
        function: Callable[..., Any] | None = None,
        *,
        firstresult: bool = False,
        historic: bool = False,
        pipeline: bool = False,
    ) -> Any:
        """Stamp ExtensionPointOpts onto the function; supports bare and called forms."""

        def mark(func: Callable[..., Any]) -> Callable[..., Any]:
            setattr(
                func, self.attribute, ExtensionPointOpts(firstresult=firstresult, historic=historic, pipeline=pipeline)
            )
            return func

        return mark(function) if function is not None else mark
Methods:
__init__(project_name)

Bind the marker to a project name used for the stamped attribute.

Source code in src/pluginkit/markers.py 
76
77
78
79
def __init__(self, project_name: str) -> None:
    """Bind the marker to a project name used for the stamped attribute."""
    self.project_name = project_name
    self.attribute = f"{project_name}_extension_point"
__call__(function=None, *, firstresult=False, historic=False, pipeline=False) 
__call__(function: Callable[P, R]) -> CollectingSpec[P, R]

__call__(
    function: None = ...,
    *,
    firstresult: Literal[True],
    historic: bool = ...,
) -> Callable[[Callable[P, R]], FirstResultSpec[P, R]]

__call__(
    function: None = ...,
    *,
    pipeline: Literal[True],
    historic: bool = ...,
) -> Callable[[Callable[P, R]], PipelineSpec[P, R]]

__call__(
    function: None = ..., *, historic: Literal[True]
) -> Callable[[Callable[P, R]], HistoricSpec[P, R]]

__call__(
    function: None = ..., *, historic: Literal[False] = ...
) -> Callable[[Callable[P, R]], CollectingSpec[P, R]]

Stamp ExtensionPointOpts onto the function; supports bare and called forms.

Source code in src/pluginkit/markers.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __call__(
    self,
    function: Callable[..., Any] | None = None,
    *,
    firstresult: bool = False,
    historic: bool = False,
    pipeline: bool = False,
) -> Any:
    """Stamp ExtensionPointOpts onto the function; supports bare and called forms."""

    def mark(func: Callable[..., Any]) -> Callable[..., Any]:
        setattr(
            func, self.attribute, ExtensionPointOpts(firstresult=firstresult, historic=historic, pipeline=pipeline)
        )
        return func

    return mark(function) if function is not None else mark

ExtensionPointOpts dataclass

Options attached to an extension point.

Source code in src/pluginkit/markers.py 
21
22
23
24
25
26
27
@dataclass(frozen=True, slots=True)
class ExtensionPointOpts:
    """Options attached to an extension point."""

    firstresult: bool = False
    historic: bool = False
    pipeline: bool = False

FirstResultSpec

A firstresult extension point: a call returns the first non-None R, or None.

Source code in src/pluginkit/markers.py 
49
50
51
52
53
54
class FirstResultSpec[**P, R]:
    """A firstresult extension point: a call returns the first non-None `R`, or `None`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
52
53
54
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

HistoricSpec

A historic extension point: replayed to late plugins, driven via call_historic.

Source code in src/pluginkit/markers.py 
65
66
67
68
69
70
class HistoricSpec[**P, R]:
    """A historic extension point: replayed to late plugins, driven via `call_historic`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
68
69
70
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

PipelineSpec

A pipeline extension point: a call threads R through the extensions and returns it.

Source code in src/pluginkit/markers.py 
57
58
59
60
61
62
class PipelineSpec[**P, R]:
    """A pipeline extension point: a call threads `R` through the extensions and returns it."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
60
61
62
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

Markers

The @extension_point / @extension decorators and the option records they stamp.

markers

Decorators that declare extension points and the extensions that fulfil them.

A marker stamps a small frozen dataclass of options onto the decorated function under a project-namespaced attribute, so the manager can later recognise extension points and extensions by introspection.

@extension_point is typed by dispatch mode: it returns a branded spec type (CollectingSpec / FirstResultSpec / PipelineSpec / HistoricSpec) that carries the call signature (P) and per-extension return type (R). PluginManager.caller reads that brand to hand back a caller whose result type is exactly right for the mode - list[R], R | None, or R. The brand classes are type-level only; they are never instantiated (an extension point is a declaration, not a callable you invoke directly).

Classes

ExtensionPointOpts dataclass

Options attached to an extension point.

Source code in src/pluginkit/markers.py 
21
22
23
24
25
26
27
@dataclass(frozen=True, slots=True)
class ExtensionPointOpts:
    """Options attached to an extension point."""

    firstresult: bool = False
    historic: bool = False
    pipeline: bool = False

ExtensionOpts dataclass

Options attached to an extension.

Source code in src/pluginkit/markers.py 
30
31
32
33
34
35
36
37
38
@dataclass(frozen=True, slots=True)
class ExtensionOpts:
    """Options attached to an extension."""

    tryfirst: bool = False
    trylast: bool = False
    wrapper: bool = False
    optional: bool = False
    target: str | None = None

CollectingSpec

A collecting extension point: a call collects each extension's R into list[R].

Source code in src/pluginkit/markers.py 
41
42
43
44
45
46
class CollectingSpec[**P, R]:
    """A collecting extension point: a call collects each extension's `R` into `list[R]`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
44
45
46
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

FirstResultSpec

A firstresult extension point: a call returns the first non-None R, or None.

Source code in src/pluginkit/markers.py 
49
50
51
52
53
54
class FirstResultSpec[**P, R]:
    """A firstresult extension point: a call returns the first non-None `R`, or `None`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
52
53
54
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

PipelineSpec

A pipeline extension point: a call threads R through the extensions and returns it.

Source code in src/pluginkit/markers.py 
57
58
59
60
61
62
class PipelineSpec[**P, R]:
    """A pipeline extension point: a call threads `R` through the extensions and returns it."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
60
61
62
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

HistoricSpec

A historic extension point: replayed to late plugins, driven via call_historic.

Source code in src/pluginkit/markers.py 
65
66
67
68
69
70
class HistoricSpec[**P, R]:
    """A historic extension point: replayed to late plugins, driven via `call_historic`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Declarations are called via PluginManager.caller(extension_point)."""
        raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")
Methods:
__call__(*args, **kwargs)

Declarations are called via PluginManager.caller(extension_point).

Source code in src/pluginkit/markers.py 
68
69
70
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Declarations are called via PluginManager.caller(extension_point)."""
    raise NotImplementedError("an extension point is a declaration; call it via PluginManager.caller(...)")

ExtensionPoint

Creates the @extension_point decorator bound to a project name.

Source code in src/pluginkit/markers.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class ExtensionPoint:
    """Creates the @extension_point decorator bound to a project name."""

    def __init__(self, project_name: str) -> None:
        """Bind the marker to a project name used for the stamped attribute."""
        self.project_name = project_name
        self.attribute = f"{project_name}_extension_point"

    @overload
    def __call__[**P, R](self, function: Callable[P, R]) -> CollectingSpec[P, R]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, firstresult: Literal[True], historic: bool = ...
    ) -> Callable[[Callable[P, R]], FirstResultSpec[P, R]]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, pipeline: Literal[True], historic: bool = ...
    ) -> Callable[[Callable[P, R]], PipelineSpec[P, R]]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, historic: Literal[True]
    ) -> Callable[[Callable[P, R]], HistoricSpec[P, R]]: ...
    @overload
    def __call__[**P, R](
        self, function: None = ..., *, historic: Literal[False] = ...
    ) -> Callable[[Callable[P, R]], CollectingSpec[P, R]]: ...
    def __call__(
        self,
        function: Callable[..., Any] | None = None,
        *,
        firstresult: bool = False,
        historic: bool = False,
        pipeline: bool = False,
    ) -> Any:
        """Stamp ExtensionPointOpts onto the function; supports bare and called forms."""

        def mark(func: Callable[..., Any]) -> Callable[..., Any]:
            setattr(
                func, self.attribute, ExtensionPointOpts(firstresult=firstresult, historic=historic, pipeline=pipeline)
            )
            return func

        return mark(function) if function is not None else mark
Methods:
__init__(project_name)

Bind the marker to a project name used for the stamped attribute.

Source code in src/pluginkit/markers.py 
76
77
78
79
def __init__(self, project_name: str) -> None:
    """Bind the marker to a project name used for the stamped attribute."""
    self.project_name = project_name
    self.attribute = f"{project_name}_extension_point"
__call__(function=None, *, firstresult=False, historic=False, pipeline=False) 
__call__(function: Callable[P, R]) -> CollectingSpec[P, R]

__call__(
    function: None = ...,
    *,
    firstresult: Literal[True],
    historic: bool = ...,
) -> Callable[[Callable[P, R]], FirstResultSpec[P, R]]

__call__(
    function: None = ...,
    *,
    pipeline: Literal[True],
    historic: bool = ...,
) -> Callable[[Callable[P, R]], PipelineSpec[P, R]]

__call__(
    function: None = ..., *, historic: Literal[True]
) -> Callable[[Callable[P, R]], HistoricSpec[P, R]]

__call__(
    function: None = ..., *, historic: Literal[False] = ...
) -> Callable[[Callable[P, R]], CollectingSpec[P, R]]

Stamp ExtensionPointOpts onto the function; supports bare and called forms.

Source code in src/pluginkit/markers.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __call__(
    self,
    function: Callable[..., Any] | None = None,
    *,
    firstresult: bool = False,
    historic: bool = False,
    pipeline: bool = False,
) -> Any:
    """Stamp ExtensionPointOpts onto the function; supports bare and called forms."""

    def mark(func: Callable[..., Any]) -> Callable[..., Any]:
        setattr(
            func, self.attribute, ExtensionPointOpts(firstresult=firstresult, historic=historic, pipeline=pipeline)
        )
        return func

    return mark(function) if function is not None else mark

Extension

Creates the @extension decorator bound to a project name.

Source code in src/pluginkit/markers.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class Extension:
    """Creates the @extension decorator bound to a project name."""

    def __init__(self, project_name: str) -> None:
        """Bind the marker to a project name used for the stamped attribute."""
        self.project_name = project_name
        self.attribute = f"{project_name}_extension"

    @overload
    def __call__[F: Callable[..., Any]](self, function: F) -> F: ...
    @overload
    def __call__[F: Callable[..., Any]](
        self,
        function: None = ...,
        *,
        tryfirst: bool = ...,
        trylast: bool = ...,
        wrapper: bool = ...,
        optional: bool = ...,
        target: str | None = ...,
    ) -> Callable[[F], F]: ...
    def __call__[F: Callable[..., Any]](
        self,
        function: F | None = None,
        *,
        tryfirst: bool = False,
        trylast: bool = False,
        wrapper: bool = False,
        optional: bool = False,
        target: str | None = None,
    ) -> F | Callable[[F], F]:
        """Stamp ExtensionOpts onto the function; supports bare and called forms."""

        def mark(func: F) -> F:
            setattr(
                func,
                self.attribute,
                ExtensionOpts(
                    tryfirst=tryfirst,
                    trylast=trylast,
                    wrapper=wrapper,
                    optional=optional,
                    target=target,
                ),
            )
            return func

        return mark(function) if function is not None else mark
Methods:
__init__(project_name)

Bind the marker to a project name used for the stamped attribute.

Source code in src/pluginkit/markers.py 
121
122
123
124
def __init__(self, project_name: str) -> None:
    """Bind the marker to a project name used for the stamped attribute."""
    self.project_name = project_name
    self.attribute = f"{project_name}_extension"
__call__(function=None, *, tryfirst=False, trylast=False, wrapper=False, optional=False, target=None) 
__call__(function: F) -> F

__call__(
    function: None = ...,
    *,
    tryfirst: bool = ...,
    trylast: bool = ...,
    wrapper: bool = ...,
    optional: bool = ...,
    target: str | None = ...,
) -> Callable[[F], F]

Stamp ExtensionOpts onto the function; supports bare and called forms.

Source code in src/pluginkit/markers.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def __call__[F: Callable[..., Any]](
    self,
    function: F | None = None,
    *,
    tryfirst: bool = False,
    trylast: bool = False,
    wrapper: bool = False,
    optional: bool = False,
    target: str | None = None,
) -> F | Callable[[F], F]:
    """Stamp ExtensionOpts onto the function; supports bare and called forms."""

    def mark(func: F) -> F:
        setattr(
            func,
            self.attribute,
            ExtensionOpts(
                tryfirst=tryfirst,
                trylast=trylast,
                wrapper=wrapper,
                optional=optional,
                target=target,
            ),
        )
        return func

    return mark(function) if function is not None else mark

Manager

The plugin manager and the dispatch internals.

manager

The plugin manager: registers plugins and dispatches calls to their hooks.

A compact but hardened reimplementation of the pluggy ideas worth understanding:

  • introspection-based discovery of specs and impls via stamped attributes;
  • registration-time validation that impl arguments exist in the spec;
  • per-impl keyword-argument filtering so an impl declares only what it needs;
  • call ordering with tryfirst / trylast;
  • collecting vs firstresult dispatch;
  • generator wrappers that decorate the result and observe exceptions safely;
  • historic hooks replayed to plugins registered later;
  • plugin lifecycle: unregister, blocking, and lookup;
  • external plugin discovery via the stdlib importlib.metadata.

The manager is safe to mutate (register / unregister / block) from multiple threads; hook calls are not internally locked and should be coordinated by the caller if they can race with registration.

Classes

HookImpl dataclass

One plugin's implementation of a hook, plus the kwargs it accepts.

Source code in src/pluginkit/manager.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(slots=True)
class HookImpl:
    """One plugin's implementation of a hook, plus the kwargs it accepts."""

    plugin_name: str
    function: Callable[..., Any]
    opts: ExtensionOpts
    accepts: frozenset[str]
    params: tuple[str, ...]
    # Set by the caller once it knows the hook's full argument set: True when this
    # impl declares exactly those arguments, so kwargs can be forwarded directly.
    passthrough: bool = False

    @classmethod
    def from_function(cls, plugin_name: str, function: Callable[..., Any], opts: ExtensionOpts) -> Self:
        """Build an impl, recording which keyword arguments the function declares."""
        params = tuple(inspect.signature(function).parameters)
        return cls(plugin_name=plugin_name, function=function, opts=opts, accepts=frozenset(params), params=params)

    def call(self, kwargs: dict[str, Any]) -> Any:
        """Invoke the function, passing only the arguments it declares.

        The caller guarantees every declared argument is present in kwargs, so the
        common "takes all the spec's arguments" case forwards kwargs directly and a
        subset impl indexes the few it wants - both avoiding a membership scan.
        """
        if self.passthrough:
            return self.function(**kwargs)
        return self.function(**{name: kwargs[name] for name in self.params})

    @property
    def order_key(self) -> int:
        """Sort key: tryfirst impls run first (0), normal next (1), trylast last (2)."""
        match self.opts:
            case ExtensionOpts(tryfirst=True):
                return 0
            case ExtensionOpts(trylast=True):
                return 2
            case _:
                return 1
Attributes
order_key property

Sort key: tryfirst impls run first (0), normal next (1), trylast last (2).

Methods:
from_function(plugin_name, function, opts) classmethod

Build an impl, recording which keyword arguments the function declares.

Source code in src/pluginkit/manager.py 
55
56
57
58
59
@classmethod
def from_function(cls, plugin_name: str, function: Callable[..., Any], opts: ExtensionOpts) -> Self:
    """Build an impl, recording which keyword arguments the function declares."""
    params = tuple(inspect.signature(function).parameters)
    return cls(plugin_name=plugin_name, function=function, opts=opts, accepts=frozenset(params), params=params)
call(kwargs)

Invoke the function, passing only the arguments it declares.

The caller guarantees every declared argument is present in kwargs, so the common "takes all the spec's arguments" case forwards kwargs directly and a subset impl indexes the few it wants - both avoiding a membership scan.

Source code in src/pluginkit/manager.py 
61
62
63
64
65
66
67
68
69
70
def call(self, kwargs: dict[str, Any]) -> Any:
    """Invoke the function, passing only the arguments it declares.

    The caller guarantees every declared argument is present in kwargs, so the
    common "takes all the spec's arguments" case forwards kwargs directly and a
    subset impl indexes the few it wants - both avoiding a membership scan.
    """
    if self.passthrough:
        return self.function(**kwargs)
    return self.function(**{name: kwargs[name] for name in self.params})

HookCaller dataclass

Holds every implementation of one hook and dispatches calls to them.

Source code in src/pluginkit/manager.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@dataclass(slots=True)
class HookCaller:
    """Holds every implementation of one hook and dispatches calls to them."""

    name: str
    spec: ExtensionPointOpts
    params: tuple[str, ...] = ()
    argnames: frozenset[str] = frozenset()
    # Default values for spec params that declare one. A call may omit these (the
    # branded caller's ParamSpec makes them optional); they are filled in at call
    # time so the type checker and the runtime agree.
    defaults: dict[str, Any] = field(default_factory=dict)
    _impls: list[HookImpl] = field(default_factory=list)
    _wrappers: list[HookImpl] = field(default_factory=list)
    _nonwrappers: list[HookImpl] = field(default_factory=list)
    _history: list[tuple[dict[str, Any], Callable[[Any], None] | None]] = field(default_factory=list)

    def __post_init__(self) -> None:
        """Derive the argument-name set from the ordered parameters when given."""
        if self.params and not self.argnames:
            self.argnames = frozenset(self.params)

    def check_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
        """Fill any omitted defaulted args, then validate the call against the spec.

        Returns the completed kwargs (defaults filled). Spec params with a default
        are optional at the call site; required params and unknown args are still
        rejected.
        """
        if self.defaults:
            kwargs = {**self.defaults, **kwargs}
        # dict_keys compares as a set against the frozenset without allocating one.
        if kwargs.keys() == self.argnames:
            return kwargs
        provided = frozenset(kwargs)
        problems: list[str] = []
        missing = self.argnames - provided
        unknown = provided - self.argnames
        if missing:
            problems.append(f"missing {sorted(missing)}")
        if unknown:
            problems.append(f"unknown {sorted(unknown)}")
        raise TypeError(f"hook {self.name!r} called with {'; '.join(problems)}; expects {sorted(self.argnames)}")

    def add_impl(self, impl: HookImpl) -> None:
        """Add an impl in priority order and replay any historic calls to it."""
        impl.passthrough = impl.accepts == self.argnames
        self._impls.append(impl)
        self._reindex()
        for kwargs, callback in self._history:
            outcome = impl.call(kwargs)
            if outcome is not None and callback is not None:
                callback(outcome)

    def remove_plugin(self, plugin_name: str) -> bool:
        """Drop every impl contributed by a plugin; return True if any were removed."""
        before = len(self._impls)
        self._impls = [impl for impl in self._impls if impl.plugin_name != plugin_name]
        removed = len(self._impls) != before
        if removed:
            self._reindex()
        return removed

    def has_plugin(self, plugin_name: str) -> bool:
        """Return whether the named plugin contributes any impl to this hook."""
        return any(impl.plugin_name == plugin_name for impl in self._impls)

    def _prepare_extra(self, functions: list[Callable[..., Any]]) -> list[HookImpl]:
        """Build one-off impls for call_extra, validating their args against the spec."""
        extra: list[HookImpl] = []
        for function in functions:
            impl = HookImpl.from_function("<call_extra>", function, ExtensionOpts())
            unknown = impl.accepts - self.argnames
            if unknown:
                raise TypeError(f"call_extra impl for {self.name!r} declares unknown argument(s) {sorted(unknown)}")
            impl.passthrough = impl.accepts == self.argnames
            extra.append(impl)
        return extra

    def _bind(self, args: tuple[Any, ...], kwargs: dict[str, Any]) -> dict[str, Any]:
        """Bind positional args to the spec's params (in order) and merge with kwargs.

        Lets a typed caller be invoked positionally - `caller(value)` as well as
        `caller(name=value)` - matching what the ParamSpec advertises.
        """
        if not args:
            return kwargs
        if len(args) > len(self.params):
            raise TypeError(f"hook {self.name!r} takes at most {len(self.params)} positional argument(s)")
        positional = dict(zip(self.params, args, strict=False))
        clash = positional.keys() & kwargs.keys()
        if clash:
            raise TypeError(f"hook {self.name!r} got multiple values for {sorted(clash)}")
        return {**positional, **kwargs}

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        """Call the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return self._execute(kwargs, self._nonwrappers)

    def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
        """Call the hook with extra one-off implementations that are not registered.

        The extra functions run as normal-priority implementations for this call
        only, ordered after the already-registered ones. Useful for tests and for
        injecting a temporary implementation without mutating the manager.
        """
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(kwargs)
        combined = sorted(
            [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
        )
        return self._execute(kwargs, combined)

    def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
        """Call a historic hook now and remember it for plugins registered later."""
        if not self.spec.historic:
            raise TypeError(f"hook {self.name!r} is not historic")
        kwargs = self.check_arguments(kwargs)
        self._history.append((kwargs, result_callback))
        for outcome in self._collect(kwargs):
            if result_callback is not None:
                result_callback(outcome)

    def _reindex(self) -> None:
        """Re-sort impls by priority and refresh the wrapper / non-wrapper split."""
        # Stable sort keeps registration order within each priority bucket.
        self._impls.sort(key=lambda candidate: candidate.order_key)
        self._wrappers = [impl for impl in self._impls if impl.opts.wrapper]
        self._nonwrappers = [impl for impl in self._impls if not impl.opts.wrapper]

    def _execute(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Run the inner impls, wrapped by any wrappers, unwinding exception-safely."""
        # Fast path: with no wrappers there is nothing to unwind, so skip the
        # try/except and generator bookkeeping entirely and let errors propagate.
        if not self._wrappers:
            return self._core(kwargs, nonwrappers)
        started: list[Generator[Any, Any, Any]] = []
        try:
            for wrapper in self._wrappers:
                generator = wrapper.call(kwargs)
                if not isinstance(generator, GeneratorType):
                    raise TypeError(f"wrapper {wrapper.plugin_name}.{self.name} must be a generator function")
                next(generator)  # advance to the yield
                started.append(generator)
            result = self._core(kwargs, nonwrappers)
        except BaseException as exc:  # noqa: BLE001 - re-raised after wrappers observe it
            return self._teardown(started, exc=exc)
        return self._teardown(started, result=result)

    def _core(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Apply the spec's dispatch strategy to the non-wrapper impls."""
        if self.spec.pipeline:
            return self._run_pipeline(kwargs, nonwrappers)
        if self.spec.firstresult:
            for impl in nonwrappers:
                outcome = impl.call(kwargs)
                if outcome is not None:
                    return outcome
            return None
        results: list[Any] = []
        for impl in nonwrappers:
            outcome = impl.call(kwargs)
            if outcome is not None:
                results.append(outcome)
        return results

    def _run_pipeline(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Thread the first argument through each impl, feeding its result to the next."""
        param = self.params[0]
        value = kwargs[param]
        current = dict(kwargs)
        for impl in nonwrappers:
            current[param] = value
            outcome = impl.call(current)
            if outcome is not None:  # None means "pass the value through unchanged"
                value = outcome
        return value

    def _collect(self, kwargs: dict[str, Any]) -> list[Any]:
        """Return the non-None results of the non-wrapper impls as a list."""
        return list(self._collect_iter(kwargs, self._nonwrappers))

    def _collect_iter(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Iterator[Any]:
        """Yield non-None results from the given non-wrapper impls, honouring firstresult."""
        for impl in nonwrappers:
            outcome = impl.call(kwargs)
            if outcome is None:
                continue
            yield outcome
            if self.spec.firstresult:
                return

    def _teardown(
        self, started: list[Generator[Any, Any, Any]], *, result: Any = _UNSET, exc: BaseException | None = None
    ) -> Any:
        """Resume each wrapper in reverse, letting it replace the result or handle the error."""
        for generator in reversed(started):
            try:
                if exc is not None:
                    generator.throw(exc)
                else:
                    generator.send(result)
            except StopIteration as stop:
                # A wrapper that returns after the yield ends here.
                if exc is not None:
                    # The wrapper swallowed the exception and supplied a result.
                    exc = None
                    result = stop.value
                elif stop.value is not None:
                    result = stop.value
            except BaseException as new_exc:  # noqa: BLE001 - propagate the wrapper's error onward
                exc = new_exc
            else:
                # The generator yielded a second time, violating the one-yield contract.
                # Capture the error but keep unwinding so the remaining wrappers still
                # tear down; the error propagates through them and is raised at the end.
                generator.close()
                exc = RuntimeError(f"wrapper for {self.name!r} must yield exactly once")
        if exc is not None:
            raise exc
        return result

    def implementations(self) -> list[HookImpl]:
        """Return this hook's implementations in call order (wrappers excluded)."""
        return list(self._nonwrappers)

    def __repr__(self) -> str:
        """Show the hook name and how many implementations it has."""
        return f"<HookCaller {self.name!r} impls={len(self._impls)}>"
Methods:
__post_init__()

Derive the argument-name set from the ordered parameters when given.

Source code in src/pluginkit/manager.py 
101
102
103
104
def __post_init__(self) -> None:
    """Derive the argument-name set from the ordered parameters when given."""
    if self.params and not self.argnames:
        self.argnames = frozenset(self.params)
check_arguments(kwargs)

Fill any omitted defaulted args, then validate the call against the spec.

Returns the completed kwargs (defaults filled). Spec params with a default are optional at the call site; required params and unknown args are still rejected.

Source code in src/pluginkit/manager.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def check_arguments(self, kwargs: dict[str, Any]) -> dict[str, Any]:
    """Fill any omitted defaulted args, then validate the call against the spec.

    Returns the completed kwargs (defaults filled). Spec params with a default
    are optional at the call site; required params and unknown args are still
    rejected.
    """
    if self.defaults:
        kwargs = {**self.defaults, **kwargs}
    # dict_keys compares as a set against the frozenset without allocating one.
    if kwargs.keys() == self.argnames:
        return kwargs
    provided = frozenset(kwargs)
    problems: list[str] = []
    missing = self.argnames - provided
    unknown = provided - self.argnames
    if missing:
        problems.append(f"missing {sorted(missing)}")
    if unknown:
        problems.append(f"unknown {sorted(unknown)}")
    raise TypeError(f"hook {self.name!r} called with {'; '.join(problems)}; expects {sorted(self.argnames)}")
add_impl(impl)

Add an impl in priority order and replay any historic calls to it.

Source code in src/pluginkit/manager.py 
128
129
130
131
132
133
134
135
136
def add_impl(self, impl: HookImpl) -> None:
    """Add an impl in priority order and replay any historic calls to it."""
    impl.passthrough = impl.accepts == self.argnames
    self._impls.append(impl)
    self._reindex()
    for kwargs, callback in self._history:
        outcome = impl.call(kwargs)
        if outcome is not None and callback is not None:
            callback(outcome)
remove_plugin(plugin_name)

Drop every impl contributed by a plugin; return True if any were removed.

Source code in src/pluginkit/manager.py 
138
139
140
141
142
143
144
145
def remove_plugin(self, plugin_name: str) -> bool:
    """Drop every impl contributed by a plugin; return True if any were removed."""
    before = len(self._impls)
    self._impls = [impl for impl in self._impls if impl.plugin_name != plugin_name]
    removed = len(self._impls) != before
    if removed:
        self._reindex()
    return removed
has_plugin(plugin_name)

Return whether the named plugin contributes any impl to this hook.

Source code in src/pluginkit/manager.py 
147
148
149
def has_plugin(self, plugin_name: str) -> bool:
    """Return whether the named plugin contributes any impl to this hook."""
    return any(impl.plugin_name == plugin_name for impl in self._impls)
__call__(*args, **kwargs)

Call the hook: a list, a single value (firstresult), or the threaded value (pipeline).

Source code in src/pluginkit/manager.py 
179
180
181
182
183
184
def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Call the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return self._execute(kwargs, self._nonwrappers)
call_extra(functions, kwargs)

Call the hook with extra one-off implementations that are not registered.

The extra functions run as normal-priority implementations for this call only, ordered after the already-registered ones. Useful for tests and for injecting a temporary implementation without mutating the manager.

Source code in src/pluginkit/manager.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
    """Call the hook with extra one-off implementations that are not registered.

    The extra functions run as normal-priority implementations for this call
    only, ordered after the already-registered ones. Useful for tests and for
    injecting a temporary implementation without mutating the manager.
    """
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(kwargs)
    combined = sorted(
        [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
    )
    return self._execute(kwargs, combined)
call_historic(kwargs, result_callback=None)

Call a historic hook now and remember it for plugins registered later.

Source code in src/pluginkit/manager.py 
201
202
203
204
205
206
207
208
209
def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
    """Call a historic hook now and remember it for plugins registered later."""
    if not self.spec.historic:
        raise TypeError(f"hook {self.name!r} is not historic")
    kwargs = self.check_arguments(kwargs)
    self._history.append((kwargs, result_callback))
    for outcome in self._collect(kwargs):
        if result_callback is not None:
            result_callback(outcome)
implementations()

Return this hook's implementations in call order (wrappers excluded).

Source code in src/pluginkit/manager.py 
310
311
312
def implementations(self) -> list[HookImpl]:
    """Return this hook's implementations in call order (wrappers excluded)."""
    return list(self._nonwrappers)
__repr__()

Show the hook name and how many implementations it has.

Source code in src/pluginkit/manager.py 
314
315
316
def __repr__(self) -> str:
    """Show the hook name and how many implementations it has."""
    return f"<HookCaller {self.name!r} impls={len(self._impls)}>"

CollectingCaller dataclass

Bases: HookCaller

A collecting hook's typed caller: a call returns list[R].

Source code in src/pluginkit/manager.py 
323
324
325
326
327
328
class CollectingCaller[**P, R](HookCaller):
    """A collecting hook's typed caller: a call returns `list[R]`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
        """Call the collecting hook, returning each impl's result as `list[R]`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Call the collecting hook, returning each impl's result as list[R].

Source code in src/pluginkit/manager.py 
326
327
328
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
    """Call the collecting hook, returning each impl's result as `list[R]`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

FirstResultCaller dataclass

Bases: HookCaller

A firstresult hook's typed caller: a call returns R | None.

Source code in src/pluginkit/manager.py 
331
332
333
334
335
336
class FirstResultCaller[**P, R](HookCaller):
    """A firstresult hook's typed caller: a call returns `R | None`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
        """Call the firstresult hook, returning the first non-None `R` or `None`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Call the firstresult hook, returning the first non-None R or None.

Source code in src/pluginkit/manager.py 
334
335
336
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
    """Call the firstresult hook, returning the first non-None `R` or `None`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

PipelineCaller dataclass

Bases: HookCaller

A pipeline hook's typed caller: a call returns R.

Source code in src/pluginkit/manager.py 
339
340
341
342
343
344
class PipelineCaller[**P, R](HookCaller):
    """A pipeline hook's typed caller: a call returns `R`."""

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Call the pipeline hook, returning the threaded value `R`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Call the pipeline hook, returning the threaded value R.

Source code in src/pluginkit/manager.py 
342
343
344
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Call the pipeline hook, returning the threaded value `R`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

HistoricCaller dataclass

Bases: HookCaller

A historic hook's typed caller. Replay it with call_historic({...}).

Calling it directly raises - historic hooks have no plain call form - so the typed __call__ is NoReturn rather than a value it never produces.

Source code in src/pluginkit/manager.py 
347
348
349
350
351
352
353
354
355
356
class HistoricCaller[**P, R](HookCaller):
    """A historic hook's typed caller. Replay it with `call_historic({...})`.

    Calling it directly raises - historic hooks have no plain call form - so the
    typed `__call__` is `NoReturn` rather than a value it never produces.
    """

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> NoReturn:
        """Historic hooks cannot be called directly; use `call_historic`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller
Methods:
__call__(*args, **kwargs)

Historic hooks cannot be called directly; use call_historic.

Source code in src/pluginkit/manager.py 
354
355
356
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> NoReturn:
    """Historic hooks cannot be called directly; use `call_historic`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

HookRelay

Attribute-style access to hook callers, e.g. pm.hook.add_ingredients(...).

Source code in src/pluginkit/manager.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
class HookRelay:
    """Attribute-style access to hook callers, e.g. pm.hook.add_ingredients(...)."""

    def __init__(self) -> None:
        """Start with no registered callers."""
        self._callers: dict[str, HookCaller] = {}

    def _add_caller(self, caller: HookCaller) -> None:
        """Register a caller under its hook name."""
        self._callers[caller.name] = caller

    def _get_caller(self, name: str) -> HookCaller | None:
        """Return the caller for a hook name, or None if undefined."""
        return self._callers.get(name)

    def _all_callers(self) -> list[HookCaller]:
        """Return every registered caller."""
        return list(self._callers.values())

    def __getattr__(self, name: str) -> HookCaller:
        """Resolve pm.hook.<name> to its HookCaller, or raise AttributeError."""
        try:
            return self._callers[name]
        except KeyError:
            raise AttributeError(f"no hook named {name!r}") from None
Methods:
__init__()

Start with no registered callers.

Source code in src/pluginkit/manager.py 
362
363
364
def __init__(self) -> None:
    """Start with no registered callers."""
    self._callers: dict[str, HookCaller] = {}
__getattr__(name)

Resolve pm.hook. to its HookCaller, or raise AttributeError.

Source code in src/pluginkit/manager.py 
378
379
380
381
382
383
def __getattr__(self, name: str) -> HookCaller:
    """Resolve pm.hook.<name> to its HookCaller, or raise AttributeError."""
    try:
        return self._callers[name]
    except KeyError:
        raise AttributeError(f"no hook named {name!r}") from None

PluginManager

Registers plugins and exposes their hooks via a HookRelay.

Source code in src/pluginkit/manager.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
class PluginManager:
    """Registers plugins and exposes their hooks via a HookRelay."""

    def __init__(self, project_name: str) -> None:
        """Bind the manager to a project name shared with the markers."""
        self.project_name = project_name
        self.hook = HookRelay()
        self._spec_attribute = f"{project_name}_extension_point"
        self._impl_attribute = f"{project_name}_extension"
        self._name2plugin: dict[str, object] = {}
        self._blocked: set[str] = set()
        self._lock = threading.RLock()

    def add_extension_points(self, namespace: object) -> None:
        """Scan a module (or object) for extension points and create callers."""
        with self._lock:
            for member_name in dir(namespace):
                member = getattr(namespace, member_name)
                spec = getattr(member, self._spec_attribute, None)
                if not isinstance(spec, ExtensionPointOpts):
                    continue
                signature = inspect.signature(member)
                params = tuple(signature.parameters)
                defaults = {
                    name: parameter.default
                    for name, parameter in signature.parameters.items()
                    if parameter.default is not inspect.Parameter.empty
                }
                self._validate_spec(member_name, spec, params)
                self.hook._add_caller(self._make_caller(member_name, spec, params, defaults))

    def _make_caller(
        self, name: str, spec: ExtensionPointOpts, params: tuple[str, ...], defaults: dict[str, Any]
    ) -> HookCaller:
        """Build the caller for a spec; overridden by AsyncPluginManager."""
        return HookCaller(name=name, spec=spec, params=params, defaults=defaults)

    @overload
    def caller[**P, R](self, spec: FirstResultSpec[P, R]) -> FirstResultCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: PipelineSpec[P, R]) -> PipelineCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: HistoricSpec[P, R]) -> HistoricCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: CollectingSpec[P, R]) -> CollectingCaller[P, R]: ...
    def caller(self, spec: object) -> HookCaller:
        """Return the typed caller for an `@extension_point`-decorated function.

        The result is a plain `HookCaller`, but its static type carries the extension
        point's dispatch mode, so a call returns `list[R]` (collecting), `R | None`
        (firstresult), or `R` (pipeline) - derived from the declaration, not asserted.
        """
        return self._caller(spec)

    def _caller(self, spec: object) -> HookCaller:
        """Resolve an extension point to its registered caller (shared by subclasses)."""
        name = getattr(spec, "__name__", None)
        if not isinstance(name, str):
            raise TypeError("caller() expects an @extension_point-decorated function")
        found = self.hook._get_caller(name)
        if found is None:
            raise PluginValidationError(
                self.project_name, f"unknown extension point {name!r}; call add_extension_points() first"
            )
        return found

    @staticmethod
    def _validate_spec(name: str, spec: ExtensionPointOpts, params: tuple[str, ...]) -> None:
        """Reject contradictory or impossible spec option combinations."""
        modes = [
            mode
            for mode, on in (
                ("firstresult", spec.firstresult),
                ("historic", spec.historic),
                ("pipeline", spec.pipeline),
            )
            if on
        ]
        if len(modes) > 1:
            raise ValueError(f"hook {name!r} cannot combine {' and '.join(modes)}")
        if spec.pipeline and not params:
            raise ValueError(f"pipeline hook {name!r} must declare at least one argument to thread through")

    def register(self, plugin: object, name: str | None = None) -> str:
        """Register a plugin object, wiring up every hook implementation it carries."""
        with self._lock:
            plugin_name = name or self.get_canonical_name(plugin)
            if plugin_name in self._blocked:
                raise ValueError(f"plugin {plugin_name!r} is blocked")
            if plugin_name in self._name2plugin:
                raise ValueError(f"plugin name {plugin_name!r} is already registered")
            if any(existing is plugin for existing in self._name2plugin.values()):
                raise ValueError(f"plugin object {plugin!r} is already registered")

            impls = self._collect_impls(plugin_name, plugin)
            self._name2plugin[plugin_name] = plugin
            try:
                for caller, impl in impls:
                    caller.add_impl(impl)
            except BaseException:
                # add_impl can fail mid-loop (e.g. a historic replay raising). Roll the
                # partial wiring back so registration is all-or-nothing.
                self._name2plugin.pop(plugin_name, None)
                for caller in self.hook._all_callers():
                    caller.remove_plugin(plugin_name)
                raise
            return plugin_name

    def unregister(self, name_or_plugin: str | object) -> object | None:
        """Remove a plugin by name or by object; return the removed plugin or None."""
        with self._lock:
            name = name_or_plugin if isinstance(name_or_plugin, str) else self.get_name(name_or_plugin)
            if name is None:
                return None
            plugin = self._name2plugin.pop(name, None)
            if plugin is None:
                return None
            for caller in self.hook._all_callers():
                caller.remove_plugin(name)
            return plugin

    def set_blocked(self, name: str) -> None:
        """Block a plugin name: unregister it if present and refuse future registration."""
        with self._lock:
            self._blocked.add(name)
            self.unregister(name)

    def is_blocked(self, name: str) -> bool:
        """Return whether a plugin name is blocked."""
        return name in self._blocked

    def is_registered(self, plugin: object) -> bool:
        """Return whether a plugin object is currently registered."""
        return any(existing is plugin for existing in self._name2plugin.values())

    def get_plugin(self, name: str) -> object | None:
        """Return the plugin registered under a name, or None."""
        return self._name2plugin.get(name)

    def get_name(self, plugin: object) -> str | None:
        """Return the registered name of a plugin object, or None."""
        for registered_name, registered_plugin in self._name2plugin.items():
            if registered_plugin is plugin:
                return registered_name
        return None

    def get_canonical_name(self, plugin: object) -> str:
        """Derive a default name for a plugin from its __name__ or type."""
        return getattr(plugin, "__name__", None) or type(plugin).__name__

    def plugin_names(self) -> list[str]:
        """Return the names of all registered plugins, in registration order."""
        return list(self._name2plugin)

    def get_hookcallers(self, plugin: object) -> list[HookCaller] | None:
        """Return the hooks a registered plugin contributes to, or None if unknown."""
        name = self.get_name(plugin)
        if name is None:
            return None
        return [caller for caller in self.hook._all_callers() if caller.has_plugin(name)]

    def __repr__(self) -> str:
        """Show the project name and number of registered plugins."""
        return f"<PluginManager {self.project_name!r} plugins={len(self._name2plugin)}>"

    def load_entrypoints(self, group: str, *, ignore_errors: bool = False) -> int:
        """Discover and register external plugins advertised under an entry-point group.

        Args:
            group: The entry-point group name to scan.
            ignore_errors: When True, skip plugins that fail to load or register
                instead of raising, so one broken plugin cannot block discovery.

        Returns:
            The number of plugins successfully registered.

        Note:
            With ``ignore_errors=False``, a failure part-way through leaves the
            plugins registered before it registered; this method does not roll back
            across plugins.
        """
        count = 0
        for entry_point in entry_points(group=group):
            if entry_point.name in self._name2plugin or entry_point.name in self._blocked:
                continue
            try:
                plugin = entry_point.load()
                self.register(plugin, name=entry_point.name)
            except Exception as error:
                if ignore_errors:
                    continue
                raise PluginValidationError(entry_point.name, f"failed to load entry point: {error}") from error
            count += 1
        return count

    def _collect_impls(self, plugin_name: str, plugin: object) -> list[tuple[HookCaller, HookImpl]]:
        """Find and validate every hook implementation a plugin carries."""
        collected: list[tuple[HookCaller, HookImpl]] = []
        for member_name in dir(plugin):
            member = getattr(plugin, member_name)
            opts = getattr(member, self._impl_attribute, None)
            if not isinstance(opts, ExtensionOpts):
                continue
            hook_name = opts.target or member_name
            caller = self.hook._get_caller(hook_name)
            if caller is None:
                if opts.optional:
                    continue
                raise PluginValidationError(plugin_name, f"implements unknown extension point {hook_name!r}")
            if opts.wrapper and caller.spec.historic:
                raise PluginValidationError(plugin_name, f"historic hook {hook_name!r} cannot have a wrapper")
            impl = HookImpl.from_function(plugin_name, member, opts)
            unknown = impl.accepts - caller.argnames
            if unknown:
                raise PluginValidationError(
                    plugin_name,
                    f"hook {hook_name!r} impl declares unknown argument(s) {sorted(unknown)}; "
                    f"spec accepts {sorted(caller.argnames)}",
                )
            collected.append((caller, impl))
        return collected
Methods:
__init__(project_name)

Bind the manager to a project name shared with the markers.

Source code in src/pluginkit/manager.py 
389
390
391
392
393
394
395
396
397
def __init__(self, project_name: str) -> None:
    """Bind the manager to a project name shared with the markers."""
    self.project_name = project_name
    self.hook = HookRelay()
    self._spec_attribute = f"{project_name}_extension_point"
    self._impl_attribute = f"{project_name}_extension"
    self._name2plugin: dict[str, object] = {}
    self._blocked: set[str] = set()
    self._lock = threading.RLock()
add_extension_points(namespace)

Scan a module (or object) for extension points and create callers.

Source code in src/pluginkit/manager.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def add_extension_points(self, namespace: object) -> None:
    """Scan a module (or object) for extension points and create callers."""
    with self._lock:
        for member_name in dir(namespace):
            member = getattr(namespace, member_name)
            spec = getattr(member, self._spec_attribute, None)
            if not isinstance(spec, ExtensionPointOpts):
                continue
            signature = inspect.signature(member)
            params = tuple(signature.parameters)
            defaults = {
                name: parameter.default
                for name, parameter in signature.parameters.items()
                if parameter.default is not inspect.Parameter.empty
            }
            self._validate_spec(member_name, spec, params)
            self.hook._add_caller(self._make_caller(member_name, spec, params, defaults))
caller(spec) 
caller(
    spec: FirstResultSpec[P, R],
) -> FirstResultCaller[P, R]

caller(spec: PipelineSpec[P, R]) -> PipelineCaller[P, R]

caller(spec: HistoricSpec[P, R]) -> HistoricCaller[P, R]

caller(
    spec: CollectingSpec[P, R],
) -> CollectingCaller[P, R]

Return the typed caller for an @extension_point-decorated function.

The result is a plain HookCaller, but its static type carries the extension point's dispatch mode, so a call returns list[R] (collecting), R | None (firstresult), or R (pipeline) - derived from the declaration, not asserted.

Source code in src/pluginkit/manager.py 
431
432
433
434
435
436
437
438
def caller(self, spec: object) -> HookCaller:
    """Return the typed caller for an `@extension_point`-decorated function.

    The result is a plain `HookCaller`, but its static type carries the extension
    point's dispatch mode, so a call returns `list[R]` (collecting), `R | None`
    (firstresult), or `R` (pipeline) - derived from the declaration, not asserted.
    """
    return self._caller(spec)
register(plugin, name=None)

Register a plugin object, wiring up every hook implementation it carries.

Source code in src/pluginkit/manager.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def register(self, plugin: object, name: str | None = None) -> str:
    """Register a plugin object, wiring up every hook implementation it carries."""
    with self._lock:
        plugin_name = name or self.get_canonical_name(plugin)
        if plugin_name in self._blocked:
            raise ValueError(f"plugin {plugin_name!r} is blocked")
        if plugin_name in self._name2plugin:
            raise ValueError(f"plugin name {plugin_name!r} is already registered")
        if any(existing is plugin for existing in self._name2plugin.values()):
            raise ValueError(f"plugin object {plugin!r} is already registered")

        impls = self._collect_impls(plugin_name, plugin)
        self._name2plugin[plugin_name] = plugin
        try:
            for caller, impl in impls:
                caller.add_impl(impl)
        except BaseException:
            # add_impl can fail mid-loop (e.g. a historic replay raising). Roll the
            # partial wiring back so registration is all-or-nothing.
            self._name2plugin.pop(plugin_name, None)
            for caller in self.hook._all_callers():
                caller.remove_plugin(plugin_name)
            raise
        return plugin_name
unregister(name_or_plugin)

Remove a plugin by name or by object; return the removed plugin or None.

Source code in src/pluginkit/manager.py 
494
495
496
497
498
499
500
501
502
503
504
505
def unregister(self, name_or_plugin: str | object) -> object | None:
    """Remove a plugin by name or by object; return the removed plugin or None."""
    with self._lock:
        name = name_or_plugin if isinstance(name_or_plugin, str) else self.get_name(name_or_plugin)
        if name is None:
            return None
        plugin = self._name2plugin.pop(name, None)
        if plugin is None:
            return None
        for caller in self.hook._all_callers():
            caller.remove_plugin(name)
        return plugin
set_blocked(name)

Block a plugin name: unregister it if present and refuse future registration.

Source code in src/pluginkit/manager.py 
507
508
509
510
511
def set_blocked(self, name: str) -> None:
    """Block a plugin name: unregister it if present and refuse future registration."""
    with self._lock:
        self._blocked.add(name)
        self.unregister(name)
is_blocked(name)

Return whether a plugin name is blocked.

Source code in src/pluginkit/manager.py 
513
514
515
def is_blocked(self, name: str) -> bool:
    """Return whether a plugin name is blocked."""
    return name in self._blocked
is_registered(plugin)

Return whether a plugin object is currently registered.

Source code in src/pluginkit/manager.py 
517
518
519
def is_registered(self, plugin: object) -> bool:
    """Return whether a plugin object is currently registered."""
    return any(existing is plugin for existing in self._name2plugin.values())
get_plugin(name)

Return the plugin registered under a name, or None.

Source code in src/pluginkit/manager.py 
521
522
523
def get_plugin(self, name: str) -> object | None:
    """Return the plugin registered under a name, or None."""
    return self._name2plugin.get(name)
get_name(plugin)

Return the registered name of a plugin object, or None.

Source code in src/pluginkit/manager.py 
525
526
527
528
529
530
def get_name(self, plugin: object) -> str | None:
    """Return the registered name of a plugin object, or None."""
    for registered_name, registered_plugin in self._name2plugin.items():
        if registered_plugin is plugin:
            return registered_name
    return None
get_canonical_name(plugin)

Derive a default name for a plugin from its name or type.

Source code in src/pluginkit/manager.py 
532
533
534
def get_canonical_name(self, plugin: object) -> str:
    """Derive a default name for a plugin from its __name__ or type."""
    return getattr(plugin, "__name__", None) or type(plugin).__name__
plugin_names()

Return the names of all registered plugins, in registration order.

Source code in src/pluginkit/manager.py 
536
537
538
def plugin_names(self) -> list[str]:
    """Return the names of all registered plugins, in registration order."""
    return list(self._name2plugin)
get_hookcallers(plugin)

Return the hooks a registered plugin contributes to, or None if unknown.

Source code in src/pluginkit/manager.py 
540
541
542
543
544
545
def get_hookcallers(self, plugin: object) -> list[HookCaller] | None:
    """Return the hooks a registered plugin contributes to, or None if unknown."""
    name = self.get_name(plugin)
    if name is None:
        return None
    return [caller for caller in self.hook._all_callers() if caller.has_plugin(name)]
__repr__()

Show the project name and number of registered plugins.

Source code in src/pluginkit/manager.py 
547
548
549
def __repr__(self) -> str:
    """Show the project name and number of registered plugins."""
    return f"<PluginManager {self.project_name!r} plugins={len(self._name2plugin)}>"
load_entrypoints(group, *, ignore_errors=False)

Discover and register external plugins advertised under an entry-point group.

Parameters:

Name Type Description Default
group str

The entry-point group name to scan.

 required
ignore_errors bool

When True, skip plugins that fail to load or register instead of raising, so one broken plugin cannot block discovery.

 False

Returns:

Type Description
int

The number of plugins successfully registered.
Note

With ignore_errors=False, a failure part-way through leaves the plugins registered before it registered; this method does not roll back across plugins.

Source code in src/pluginkit/manager.py 
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
def load_entrypoints(self, group: str, *, ignore_errors: bool = False) -> int:
    """Discover and register external plugins advertised under an entry-point group.

    Args:
        group: The entry-point group name to scan.
        ignore_errors: When True, skip plugins that fail to load or register
            instead of raising, so one broken plugin cannot block discovery.

    Returns:
        The number of plugins successfully registered.

    Note:
        With ``ignore_errors=False``, a failure part-way through leaves the
        plugins registered before it registered; this method does not roll back
        across plugins.
    """
    count = 0
    for entry_point in entry_points(group=group):
        if entry_point.name in self._name2plugin or entry_point.name in self._blocked:
            continue
        try:
            plugin = entry_point.load()
            self.register(plugin, name=entry_point.name)
        except Exception as error:
            if ignore_errors:
                continue
            raise PluginValidationError(entry_point.name, f"failed to load entry point: {error}") from error
        count += 1
    return count

Async

The async manager and hook caller.

aio

Async dispatch: an AsyncPluginManager that awaits coroutine implementations.

The registration, validation and lifecycle machinery is reused unchanged from the synchronous manager; only the calling path is asynchronous. Implementations may be plain functions or coroutine functions - their results are awaited when awaitable. Collecting, firstresult and pipeline dispatch are all supported.

Wrappers in the async manager are async generators and are observe-only: they run setup before yield and teardown after it (including in a finally), and they observe exceptions thrown back in, but - because async generators cannot return a value - they do not replace the result. Use the synchronous manager when a wrapper must transform the result.

Classes

AsyncHookCaller dataclass

Bases: HookCaller

A HookCaller whose calls are coroutines that await async implementations.

Source code in src/pluginkit/aio.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class AsyncHookCaller(HookCaller):
    """A HookCaller whose calls are coroutines that await async implementations."""

    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
        """Await the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return await self._execute_async(kwargs, self._nonwrappers)

    async def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
        """Await the hook with extra one-off implementations that are not registered."""
        if self.spec.historic:
            raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
        kwargs = self.check_arguments(kwargs)
        combined = sorted(
            [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
        )
        return await self._execute_async(kwargs, combined)

    def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
        """Historic hooks are not supported by the async manager."""
        raise NotImplementedError("historic hooks are not supported by AsyncPluginManager")

    async def _execute_async(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Start async wrappers, run the impls, then unwind wrappers exception-safely."""
        started: list[AsyncGeneratorType[Any, Any]] = []
        try:
            for wrapper in self._wrappers:
                generator = wrapper.call(kwargs)
                if not isinstance(generator, AsyncGeneratorType):
                    raise TypeError(f"async wrapper {wrapper.plugin_name}.{self.name} must be an async generator")
                await generator.__anext__()  # advance to the yield
                started.append(generator)
            result = await self._core_async(kwargs, nonwrappers)
        except BaseException as exc:  # noqa: BLE001 - re-raised after wrappers observe it
            return await self._teardown_async(started, exc=exc)
        return await self._teardown_async(started, result=result)

    async def _core_async(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Apply the spec's dispatch strategy, awaiting any awaitable results."""
        if self.spec.pipeline:
            return await self._run_pipeline_async(kwargs, nonwrappers)
        results: list[Any] = []
        for impl in nonwrappers:
            outcome = await _maybe_await(impl.call(kwargs))
            if outcome is None:
                continue
            results.append(outcome)
            if self.spec.firstresult:
                break
        return (results[0] if results else None) if self.spec.firstresult else results

    async def _run_pipeline_async(self, kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> Any:
        """Thread the first argument through each impl, awaiting awaitable results."""
        param = self.params[0]
        value = kwargs[param]
        current = dict(kwargs)
        for impl in nonwrappers:
            current[param] = value
            outcome = await _maybe_await(impl.call(current))
            if outcome is not None:
                value = outcome
        return value

    async def _teardown_async(
        self, started: list[AsyncGeneratorType[Any, Any]], *, result: Any = _UNSET, exc: BaseException | None = None
    ) -> Any:
        """Resume each async wrapper in reverse so its teardown runs and it observes errors."""
        for generator in reversed(started):
            try:
                if exc is not None:
                    await generator.athrow(exc)
                else:
                    await generator.asend(result)
            except StopAsyncIteration:
                pass  # normal completion; async wrappers cannot replace the result
            except BaseException as new_exc:  # noqa: BLE001 - propagate the wrapper's error onward
                exc = new_exc
            else:
                # Double yield: capture the error but keep unwinding the remaining
                # wrappers so their teardown still runs; raised after the loop.
                await generator.aclose()
                exc = RuntimeError(f"async wrapper for {self.name!r} must yield exactly once")
        if exc is not None:
            raise exc
        return result
Methods:
__call__(*args, **kwargs) async

Await the hook: a list, a single value (firstresult), or the threaded value (pipeline).

Source code in src/pluginkit/aio.py 
32
33
34
35
36
37
async def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Await the hook: a list, a single value (firstresult), or the threaded value (pipeline)."""
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return await self._execute_async(kwargs, self._nonwrappers)
call_extra(functions, kwargs) async

Await the hook with extra one-off implementations that are not registered.

Source code in src/pluginkit/aio.py 
39
40
41
42
43
44
45
46
47
async def call_extra(self, functions: list[Callable[..., Any]], kwargs: dict[str, Any]) -> Any:
    """Await the hook with extra one-off implementations that are not registered."""
    if self.spec.historic:
        raise TypeError(f"historic hook {self.name!r} must be called via call_historic()")
    kwargs = self.check_arguments(kwargs)
    combined = sorted(
        [*self._nonwrappers, *self._prepare_extra(functions)], key=lambda candidate: candidate.order_key
    )
    return await self._execute_async(kwargs, combined)
call_historic(kwargs, result_callback=None)

Historic hooks are not supported by the async manager.

Source code in src/pluginkit/aio.py 
49
50
51
def call_historic(self, kwargs: dict[str, Any], result_callback: Callable[[Any], None] | None = None) -> None:
    """Historic hooks are not supported by the async manager."""
    raise NotImplementedError("historic hooks are not supported by AsyncPluginManager")

AsyncCollectingCaller dataclass

Bases: AsyncHookCaller

A collecting async hook's typed caller: await a call to get list[R].

Source code in src/pluginkit/aio.py 
120
121
122
123
124
125
class AsyncCollectingCaller[**P, R](AsyncHookCaller):
    """A collecting async hook's typed caller: `await` a call to get `list[R]`."""

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
        """Await the collecting hook, returning each impl's result as `list[R]`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller
Methods:
__call__(*args, **kwargs) async

Await the collecting hook, returning each impl's result as list[R].

Source code in src/pluginkit/aio.py 
123
124
125
async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> list[R]:
    """Await the collecting hook, returning each impl's result as `list[R]`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller

AsyncFirstResultCaller dataclass

Bases: AsyncHookCaller

A firstresult async hook's typed caller: await a call to get R | None.

Source code in src/pluginkit/aio.py 
128
129
130
131
132
133
class AsyncFirstResultCaller[**P, R](AsyncHookCaller):
    """A firstresult async hook's typed caller: `await` a call to get `R | None`."""

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
        """Await the firstresult hook, returning the first non-None `R` or `None`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller
Methods:
__call__(*args, **kwargs) async

Await the firstresult hook, returning the first non-None R or None.

Source code in src/pluginkit/aio.py 
131
132
133
async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R | None:
    """Await the firstresult hook, returning the first non-None `R` or `None`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller

AsyncPipelineCaller dataclass

Bases: AsyncHookCaller

A pipeline async hook's typed caller: await a call to get R.

Source code in src/pluginkit/aio.py 
136
137
138
139
140
141
class AsyncPipelineCaller[**P, R](AsyncHookCaller):
    """A pipeline async hook's typed caller: `await` a call to get `R`."""

    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Await the pipeline hook, returning the threaded value `R`."""
        raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller
Methods:
__call__(*args, **kwargs) async

Await the pipeline hook, returning the threaded value R.

Source code in src/pluginkit/aio.py 
139
140
141
async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Await the pipeline hook, returning the threaded value `R`."""
    raise NotImplementedError  # pragma: no cover - the runtime object is an AsyncHookCaller

AsyncPluginManager

Bases: PluginManager

A PluginManager whose hooks are awaited; impls may be coroutine functions.

Source code in src/pluginkit/aio.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class AsyncPluginManager(PluginManager):
    """A PluginManager whose hooks are awaited; impls may be coroutine functions."""

    def _make_caller(
        self, name: str, spec: ExtensionPointOpts, params: tuple[str, ...], defaults: dict[str, Any]
    ) -> HookCaller:
        """Build an AsyncHookCaller instead of the synchronous one."""
        return AsyncHookCaller(name=name, spec=spec, params=params, defaults=defaults)

    @overload  # type: ignore[override]  # async manager returns awaitable callers
    def caller[**P, R](self, spec: FirstResultSpec[P, R]) -> AsyncFirstResultCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: PipelineSpec[P, R]) -> AsyncPipelineCaller[P, R]: ...
    @overload
    def caller[**P, R](self, spec: CollectingSpec[P, R]) -> AsyncCollectingCaller[P, R]: ...
    def caller(  # pyright: ignore[reportIncompatibleMethodOverride]  # async returns awaitable callers
        self, spec: object
    ) -> HookCaller:
        """Return the typed async caller for an `@extension_point`-decorated function."""
        return self._caller(spec)
Methods:
caller(spec) 
caller(
    spec: FirstResultSpec[P, R],
) -> AsyncFirstResultCaller[P, R]

caller(
    spec: PipelineSpec[P, R],
) -> AsyncPipelineCaller[P, R]

caller(
    spec: CollectingSpec[P, R],
) -> AsyncCollectingCaller[P, R]

Return the typed async caller for an @extension_point-decorated function.

Source code in src/pluginkit/aio.py 
159
160
161
162
163
def caller(  # pyright: ignore[reportIncompatibleMethodOverride]  # async returns awaitable callers
    self, spec: object
) -> HookCaller:
    """Return the typed async caller for an `@extension_point`-decorated function."""
    return self._caller(spec)

Exceptions

exceptions

Exceptions raised by the plugin framework.

Classes

PluginValidationError

Bases: Exception

Raised when a plugin or one of its hook implementations is invalid.

Source code in src/pluginkit/exceptions.py 
 4
 5
 6
 7
 8
 9
10
class PluginValidationError(Exception):
    """Raised when a plugin or one of its hook implementations is invalid."""

    def __init__(self, plugin_name: str, message: str) -> None:
        """Record the offending plugin name alongside the message."""
        self.plugin_name = plugin_name
        super().__init__(f"plugin {plugin_name!r}: {message}")
Methods:
__init__(plugin_name, message)

Record the offending plugin name alongside the message.

Source code in src/pluginkit/exceptions.py 
 7
 8
 9
10
def __init__(self, plugin_name: str, message: str) -> None:
    """Record the offending plugin name alongside the message."""
    self.plugin_name = plugin_name
    super().__init__(f"plugin {plugin_name!r}: {message}")

Demo host

The bundled "smoothie kitchen" host that the walkthrough demos use.

Extension points

points

Hook specifications for the host, plus a Protocol documenting the contract.

The @extension_point functions are what the manager discovers. The Protocol classes are a modern-Python touch: they let a type checker verify, structurally, that a plugin's method signatures line up with the hooks they implement -- no base class or inheritance required.

Classes

IngredientProvider

Bases: Protocol

Structural type of any plugin that contributes ingredients.

Source code in examples/tour/src/pluginkit_tour/points.py 
43
44
45
46
47
48
49
@runtime_checkable
class IngredientProvider(Protocol):
    """Structural type of any plugin that contributes ingredients."""

    def add_ingredients(self, base: list[str]) -> list[str]:
        """Return ingredients to add to the smoothie."""
        ...
Methods:
add_ingredients(base)

Return ingredients to add to the smoothie.

Source code in examples/tour/src/pluginkit_tour/points.py 
47
48
49
def add_ingredients(self, base: list[str]) -> list[str]:
    """Return ingredients to add to the smoothie."""
    ...

Functions:

add_ingredients(base)

Offer ingredients to add; results are collected into a list.

Source code in examples/tour/src/pluginkit_tour/points.py 
18
19
20
@extension_point
def add_ingredients(base: list[str]) -> list[str]:
    """Offer ingredients to add; results are collected into a list."""

choose_cup(size)

Pick a cup for the size; the first plugin to answer (non-None) wins.

Source code in examples/tour/src/pluginkit_tour/points.py 
23
24
25
@extension_point(firstresult=True)
def choose_cup(size: str) -> str | None:
    """Pick a cup for the size; the first plugin to answer (non-None) wins."""

prep_step(steps)

Append a preparation step in place; cross-plugin ordering is the point.

Source code in examples/tour/src/pluginkit_tour/points.py 
28
29
30
@extension_point
def prep_step(steps: list[str]) -> None:
    """Append a preparation step in place; cross-plugin ordering is the point."""

blend(contents)

Blend the contents into one drink; wrappers decorate that single result.

Source code in examples/tour/src/pluginkit_tour/points.py 
33
34
35
@extension_point(firstresult=True)
def blend(contents: list[str]) -> str | None:
    """Blend the contents into one drink; wrappers decorate that single result."""

kitchen_opened(name)

Announce the kitchen opened; late-registered plugins still hear it.

Source code in examples/tour/src/pluginkit_tour/points.py 
38
39
40
@extension_point(historic=True)
def kitchen_opened(name: str) -> None:
    """Announce the kitchen opened; late-registered plugins still hear it."""