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:PluginResult / :class:EntryPointLoadReport - structured provenance and discovery outcomes for production hosts.
  • :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 - public low-level dispatch objects for inspection and advanced integrations.
  • :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 
154
155
156
157
158
159
160
161
162
163
164
165
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

    async def collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
        self, *args: P.args, **kwargs: P.kwargs
    ) -> list[PluginResult[R]]:
        """Await results paired with the plugin names that produced them."""
        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 
157
158
159
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
collect_with_plugins(*args, **kwargs) async

Await results paired with the plugin names that produced them.

Source code in src/pluginkit/aio.py 
161
162
163
164
165
async def collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
    self, *args: P.args, **kwargs: P.kwargs
) -> list[PluginResult[R]]:
    """Await results paired with the plugin names that produced them."""
    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 
168
169
170
171
172
173
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 
171
172
173
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
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
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 collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
        self, *args: Any, **kwargs: Any
    ) -> list[PluginResult[Any]]:
        """Await collecting implementations and preserve each result's plugin name."""
        if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
            raise TypeError(f"hook {self.name!r} is not a collecting hook")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return cast(list[PluginResult[Any]], await self._execute_async(kwargs, self._nonwrappers, with_plugins=True))

    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], *, with_plugins: bool = False
    ) -> 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")
                try:
                    await generator.__anext__()  # advance to the yield
                except StopAsyncIteration:
                    raise RuntimeError(f"async wrapper for {self.name!r} must yield exactly once") from None
                started.append(generator)
            result = (
                await self._core_with_plugins_async(kwargs, nonwrappers)
                if with_plugins
                else 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

    @staticmethod
    async def _core_with_plugins_async(kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> list[PluginResult[Any]]:
        """Await and collect non-None results with provenance."""
        results: list[PluginResult[Any]] = []
        for impl in nonwrappers:
            outcome = await _maybe_await(impl.call(kwargs))
            if outcome is not None:
                results.append(PluginResult(impl.plugin_name, outcome))
        return 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. A failing
                # aclose() carries the contract violation as its cause and does not stop
                # the unwind either.
                exc = RuntimeError(f"async wrapper for {self.name!r} must yield exactly once")
                try:
                    await generator.aclose()
                except BaseException as close_exc:  # noqa: BLE001 - keep unwinding the rest
                    close_exc.__cause__ = exc
                    exc = close_exc
        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)
collect_with_plugins(*args, **kwargs) async

Await collecting implementations and preserve each result's plugin name.

Source code in src/pluginkit/aio.py 
39
40
41
42
43
44
45
46
async def collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
    self, *args: Any, **kwargs: Any
) -> list[PluginResult[Any]]:
    """Await collecting implementations and preserve each result's plugin name."""
    if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
        raise TypeError(f"hook {self.name!r} is not a collecting hook")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return cast(list[PluginResult[Any]], await self._execute_async(kwargs, self._nonwrappers, with_plugins=True))
call_extra(functions, kwargs) async

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

Source code in src/pluginkit/aio.py 
48
49
50
51
52
53
54
55
56
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 
58
59
60
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 
176
177
178
179
180
181
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 
179
180
181
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 
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
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], arity: int
    ) -> HookCaller:
        """Build an AsyncHookCaller; historic extension points are not supported async-side."""
        if spec.historic:
            raise ValueError(
                f"historic extension point {name!r} is not supported by AsyncPluginManager "
                f"(historic replay requires synchronous dispatch)"
            )
        return AsyncHookCaller(name=name, spec=spec, params=params, defaults=defaults, positional_arity=arity)

    @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 
204
205
206
207
208
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 
413
414
415
416
417
418
419
420
421
422
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

    def collect_with_plugins(self, *args: P.args, **kwargs: P.kwargs) -> list[PluginResult[R]]:
        """Collect results paired with the plugin names that produced them."""
        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 
416
417
418
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
collect_with_plugins(*args, **kwargs)

Collect results paired with the plugin names that produced them.

Source code in src/pluginkit/manager.py 
420
421
422
def collect_with_plugins(self, *args: P.args, **kwargs: P.kwargs) -> list[PluginResult[R]]:
    """Collect results paired with the plugin names that produced them."""
    raise NotImplementedError  # pragma: no cover - the runtime object is a HookCaller

EntryPointFailure

Bases: NamedTuple

An entry point that could not be loaded or registered.

Source code in src/pluginkit/manager.py 
117
118
119
120
121
class EntryPointFailure(NamedTuple):
    """An entry point that could not be loaded or registered."""

    name: str
    error: Exception

EntryPointLoadReport dataclass

Structured outcome of resilient entry-point discovery.

Source code in src/pluginkit/manager.py 
124
125
126
127
128
129
130
131
@dataclass(frozen=True, slots=True)
class EntryPointLoadReport:
    """Structured outcome of resilient entry-point discovery."""

    loaded: tuple[str, ...]
    already_loaded: tuple[str, ...]
    blocked: tuple[str, ...]
    failed: tuple[EntryPointFailure, ...]

FirstResultCaller dataclass

Bases: HookCaller

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

Source code in src/pluginkit/manager.py 
425
426
427
428
429
430
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 
428
429
430
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 
441
442
443
444
445
446
447
448
449
450
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 
448
449
450
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 
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
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
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
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
@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)
    # How many leading params may be passed positionally (the rest are keyword-only);
    # -1 means "derive as all params" for callers built directly without kind info.
    positional_arity: int = -1
    _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)
        if self.positional_arity < 0:
            self.positional_arity = len(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:
            try:
                impl = HookImpl.from_function("<call_extra>", function, ExtensionOpts())
            except ValueError as exc:
                raise TypeError(str(exc)) from exc
            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) > self.positional_arity:
            raise TypeError(f"hook {self.name!r} takes at most {self.positional_arity} positional argument(s)")
        positional = dict(zip(self.params[: self.positional_arity], 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 collect_with_plugins(self, *args: Any, **kwargs: Any) -> list[PluginResult[Any]]:
        """Collect non-None results together with their registered plugin names."""
        if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
            raise TypeError(f"hook {self.name!r} is not a collecting hook")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return cast(list[PluginResult[Any]], self._execute(kwargs, self._nonwrappers, with_plugins=True))

    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)
        # Dispatch to the current impls and fire their callbacks first: if either the
        # dispatch or a callback raises, the call fails without recording the event, so a
        # later plugin never replays (and re-fails on) a call that did not fully succeed.
        outcomes = self._collect(kwargs)
        for outcome in outcomes:
            if result_callback is not None:
                result_callback(outcome)
        # Record only after a fully successful call, snapshotting the event so a later
        # mutation of the caller's dict can't change what plugins replay.
        self._history.append((dict(kwargs), result_callback))

    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], *, with_plugins: bool = False) -> 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_with_plugins(kwargs, nonwrappers) if with_plugins else 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")
                try:
                    next(generator)  # advance to the yield
                except StopIteration:
                    raise RuntimeError(f"wrapper for {self.name!r} must yield exactly once") from None
                started.append(generator)
            result = self._core_with_plugins(kwargs, nonwrappers) if with_plugins else 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

    @staticmethod
    def _core_with_plugins(kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> list[PluginResult[Any]]:
        """Collect non-None results with provenance in ordinary dispatch order."""
        results: list[PluginResult[Any]] = []
        for impl in nonwrappers:
            outcome = impl.call(kwargs)
            if outcome is not None:
                results.append(PluginResult(impl.plugin_name, 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.
                # Closing the offender can itself raise (a failing `finally`); that error
                # then carries the contract violation as its cause and unwinding continues.
                exc = RuntimeError(f"wrapper for {self.name!r} must yield exactly once")
                try:
                    generator.close()
                except BaseException as close_exc:  # noqa: BLE001 - keep unwinding the rest
                    close_exc.__cause__ = exc
                    exc = close_exc
        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 
154
155
156
157
158
159
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)
    if self.positional_arity < 0:
        self.positional_arity = len(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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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 
183
184
185
186
187
188
189
190
191
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 
193
194
195
196
197
198
199
200
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 
202
203
204
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 
237
238
239
240
241
242
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)
collect_with_plugins(*args, **kwargs)

Collect non-None results together with their registered plugin names.

Source code in src/pluginkit/manager.py 
244
245
246
247
248
249
def collect_with_plugins(self, *args: Any, **kwargs: Any) -> list[PluginResult[Any]]:
    """Collect non-None results together with their registered plugin names."""
    if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
        raise TypeError(f"hook {self.name!r} is not a collecting hook")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return cast(list[PluginResult[Any]], self._execute(kwargs, self._nonwrappers, with_plugins=True))
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 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
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 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
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)
    # Dispatch to the current impls and fire their callbacks first: if either the
    # dispatch or a callback raises, the call fails without recording the event, so a
    # later plugin never replays (and re-fails on) a call that did not fully succeed.
    outcomes = self._collect(kwargs)
    for outcome in outcomes:
        if result_callback is not None:
            result_callback(outcome)
    # Record only after a fully successful call, snapshotting the event so a later
    # mutation of the caller's dict can't change what plugins replay.
    self._history.append((dict(kwargs), result_callback))
implementations()

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

Source code in src/pluginkit/manager.py 
400
401
402
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 
404
405
406
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 
 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
@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."""
        signature = inspect.signature(function)
        _positional_arity(signature, "implementation", getattr(function, "__qualname__", plugin_name))
        params = tuple(signature.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 
79
80
81
82
83
84
85
@classmethod
def from_function(cls, plugin_name: str, function: Callable[..., Any], opts: ExtensionOpts) -> Self:
    """Build an impl, recording which keyword arguments the function declares."""
    signature = inspect.signature(function)
    _positional_arity(signature, "implementation", getattr(function, "__qualname__", plugin_name))
    params = tuple(signature.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 
87
88
89
90
91
92
93
94
95
96
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 
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
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 
456
457
458
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 
472
473
474
475
476
477
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 
433
434
435
436
437
438
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 
436
437
438
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 
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
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
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._spec_callers: dict[int, tuple[object, HookCaller]] = {}
        self._lock = threading.RLock()

    def add_extension_points(self, namespace: object) -> None:
        """Scan a module (or object) for extension points and create callers.

        Every caller is built and validated before any is registered, so a namespace
        with an invalid spec leaves the manager unchanged rather than half-configured.
        """
        with self._lock:
            new_callers: list[HookCaller] = []
            new_specs: list[tuple[object, HookCaller]] = []
            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)
                arity = _positional_arity(signature, "extension point", member_name)
                params = tuple(signature.parameters)
                defaults = {
                    name: parameter.default
                    for name, parameter in signature.parameters.items()
                    if parameter.default is not inspect.Parameter.empty
                }
                if self.hook._get_caller(member_name) is not None:
                    raise ValueError(
                        f"extension point {member_name!r} is already registered; "
                        f"re-adding it would drop the implementations already wired to it"
                    )
                self._validate_spec(member_name, spec, signature)
                caller = self._make_caller(member_name, spec, params, defaults, arity)
                new_callers.append(caller)
                new_specs.append((self._spec_identity(member), caller))
            for caller in new_callers:
                self.hook._add_caller(caller)
            for member, caller in new_specs:
                self._spec_callers[id(member)] = (member, caller)

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

    @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)."""
        identity = self._spec_identity(spec)
        name = getattr(spec, "__name__", None)
        if not isinstance(name, str):
            raise TypeError("caller() expects an @extension_point-decorated function")
        marker = getattr(spec, self._spec_attribute, None)
        if not isinstance(marker, ExtensionPointOpts):
            raise TypeError(f"caller() expects an extension point decorated for project {self.project_name!r}")
        registered = self._spec_callers.get(id(identity))
        if registered is None or registered[0] is not identity:
            raise PluginValidationError(
                self.project_name, f"unknown extension point {name!r}; call add_extension_points() first"
            )
        return registered[1]

    @staticmethod
    def _spec_identity(spec: object) -> object:
        """Normalize bound methods to their stable underlying function identity."""
        return getattr(spec, "__func__", spec)

    @staticmethod
    def _validate_spec(name: str, spec: ExtensionPointOpts, signature: inspect.Signature) -> None:
        """Reject contradictory or impossible spec option combinations."""
        params = tuple(signature.parameters)
        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")
        if spec.pipeline:
            threaded = signature.parameters[params[0]].annotation
            returned = signature.return_annotation
            if _is_concrete_annotation(threaded) and _is_concrete_annotation(returned) and threaded != returned:
                raise ValueError(
                    f"pipeline hook {name!r} must return its threaded first argument type; "
                    f"got {threaded!r} -> {returned!r}"
                )

    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:
                    if self._name2plugin.get(plugin_name) is not plugin:
                        # A historic replay unregistered or blocked the plugin from inside
                        # one of its own implementations. Stop wiring the rest so no hook
                        # outlives the registration it belonged to.
                        for hook_caller in self.hook._all_callers():
                            hook_caller.remove_plugin(plugin_name)
                        return plugin_name
                    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 load_entrypoints_report(self, group: str) -> EntryPointLoadReport:
        """Discover entry points resiliently and return structured outcomes.

        Unlike strict :meth:`load_entrypoints`, this method always continues after
        individual failures. The original exception is retained for host reporting.
        """
        loaded: list[str] = []
        already_loaded: list[str] = []
        blocked: list[str] = []
        failed: list[EntryPointFailure] = []
        for entry_point in entry_points(group=group):
            if entry_point.name in self._name2plugin:
                already_loaded.append(entry_point.name)
                continue
            if entry_point.name in self._blocked:
                blocked.append(entry_point.name)
                continue
            try:
                plugin = entry_point.load()
                self.register(plugin, name=entry_point.name)
            except Exception as error:
                failed.append(EntryPointFailure(entry_point.name, error))
            else:
                loaded.append(entry_point.name)
        return EntryPointLoadReport(
            loaded=tuple(loaded),
            already_loaded=tuple(already_loaded),
            blocked=tuple(blocked),
            failed=tuple(failed),
        )

    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")
            if opts.tryfirst and opts.trylast:
                raise PluginValidationError(
                    plugin_name, f"hook {hook_name!r} cannot combine tryfirst=True and trylast=True"
                )
            try:
                impl = HookImpl.from_function(plugin_name, member, opts)
            except ValueError as exc:
                raise PluginValidationError(plugin_name, str(exc)) from exc
            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 
483
484
485
486
487
488
489
490
491
492
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._spec_callers: dict[int, tuple[object, HookCaller]] = {}
    self._lock = threading.RLock()
add_extension_points(namespace)

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

Every caller is built and validated before any is registered, so a namespace with an invalid spec leaves the manager unchanged rather than half-configured.

Source code in src/pluginkit/manager.py 
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
def add_extension_points(self, namespace: object) -> None:
    """Scan a module (or object) for extension points and create callers.

    Every caller is built and validated before any is registered, so a namespace
    with an invalid spec leaves the manager unchanged rather than half-configured.
    """
    with self._lock:
        new_callers: list[HookCaller] = []
        new_specs: list[tuple[object, HookCaller]] = []
        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)
            arity = _positional_arity(signature, "extension point", member_name)
            params = tuple(signature.parameters)
            defaults = {
                name: parameter.default
                for name, parameter in signature.parameters.items()
                if parameter.default is not inspect.Parameter.empty
            }
            if self.hook._get_caller(member_name) is not None:
                raise ValueError(
                    f"extension point {member_name!r} is already registered; "
                    f"re-adding it would drop the implementations already wired to it"
                )
            self._validate_spec(member_name, spec, signature)
            caller = self._make_caller(member_name, spec, params, defaults, arity)
            new_callers.append(caller)
            new_specs.append((self._spec_identity(member), caller))
        for caller in new_callers:
            self.hook._add_caller(caller)
        for member, caller in new_specs:
            self._spec_callers[id(member)] = (member, caller)
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 
544
545
546
547
548
549
550
551
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 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
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:
                if self._name2plugin.get(plugin_name) is not plugin:
                    # A historic replay unregistered or blocked the plugin from inside
                    # one of its own implementations. Stop wiring the rest so no hook
                    # outlives the registration it belonged to.
                    for hook_caller in self.hook._all_callers():
                        hook_caller.remove_plugin(plugin_name)
                    return plugin_name
                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 
632
633
634
635
636
637
638
639
640
641
642
643
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 
645
646
647
648
649
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 
651
652
653
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 
655
656
657
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 
659
660
661
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 
663
664
665
666
667
668
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 
670
671
672
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 
674
675
676
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 
678
679
680
681
682
683
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 
685
686
687
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 
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
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
load_entrypoints_report(group)

Discover entry points resiliently and return structured outcomes.

Unlike strict :meth:load_entrypoints, this method always continues after individual failures. The original exception is retained for host reporting.

Source code in src/pluginkit/manager.py 
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
def load_entrypoints_report(self, group: str) -> EntryPointLoadReport:
    """Discover entry points resiliently and return structured outcomes.

    Unlike strict :meth:`load_entrypoints`, this method always continues after
    individual failures. The original exception is retained for host reporting.
    """
    loaded: list[str] = []
    already_loaded: list[str] = []
    blocked: list[str] = []
    failed: list[EntryPointFailure] = []
    for entry_point in entry_points(group=group):
        if entry_point.name in self._name2plugin:
            already_loaded.append(entry_point.name)
            continue
        if entry_point.name in self._blocked:
            blocked.append(entry_point.name)
            continue
        try:
            plugin = entry_point.load()
            self.register(plugin, name=entry_point.name)
        except Exception as error:
            failed.append(EntryPointFailure(entry_point.name, error))
        else:
            loaded.append(entry_point.name)
    return EntryPointLoadReport(
        loaded=tuple(loaded),
        already_loaded=tuple(already_loaded),
        blocked=tuple(blocked),
        failed=tuple(failed),
    )

PluginResult

Bases: NamedTuple

One collecting-hook result paired with the plugin that produced it.

Source code in src/pluginkit/manager.py 
110
111
112
113
114
class PluginResult[R](NamedTuple):
    """One collecting-hook result paired with the plugin that produced it."""

    plugin_name: str
    value: R

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, structured discovery and attribution records, and public low-level dispatch objects. Prefer manager and caller methods for ordinary host code; mode-specific caller subclasses are typing facades rather than runtime classes.

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 
 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
@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."""
        signature = inspect.signature(function)
        _positional_arity(signature, "implementation", getattr(function, "__qualname__", plugin_name))
        params = tuple(signature.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 
79
80
81
82
83
84
85
@classmethod
def from_function(cls, plugin_name: str, function: Callable[..., Any], opts: ExtensionOpts) -> Self:
    """Build an impl, recording which keyword arguments the function declares."""
    signature = inspect.signature(function)
    _positional_arity(signature, "implementation", getattr(function, "__qualname__", plugin_name))
    params = tuple(signature.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 
87
88
89
90
91
92
93
94
95
96
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})

PluginResult

Bases: NamedTuple

One collecting-hook result paired with the plugin that produced it.

Source code in src/pluginkit/manager.py 
110
111
112
113
114
class PluginResult[R](NamedTuple):
    """One collecting-hook result paired with the plugin that produced it."""

    plugin_name: str
    value: R

EntryPointFailure

Bases: NamedTuple

An entry point that could not be loaded or registered.

Source code in src/pluginkit/manager.py 
117
118
119
120
121
class EntryPointFailure(NamedTuple):
    """An entry point that could not be loaded or registered."""

    name: str
    error: Exception

EntryPointLoadReport dataclass

Structured outcome of resilient entry-point discovery.

Source code in src/pluginkit/manager.py 
124
125
126
127
128
129
130
131
@dataclass(frozen=True, slots=True)
class EntryPointLoadReport:
    """Structured outcome of resilient entry-point discovery."""

    loaded: tuple[str, ...]
    already_loaded: tuple[str, ...]
    blocked: tuple[str, ...]
    failed: tuple[EntryPointFailure, ...]

HookCaller dataclass

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

Source code in src/pluginkit/manager.py 
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
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
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
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
@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)
    # How many leading params may be passed positionally (the rest are keyword-only);
    # -1 means "derive as all params" for callers built directly without kind info.
    positional_arity: int = -1
    _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)
        if self.positional_arity < 0:
            self.positional_arity = len(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:
            try:
                impl = HookImpl.from_function("<call_extra>", function, ExtensionOpts())
            except ValueError as exc:
                raise TypeError(str(exc)) from exc
            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) > self.positional_arity:
            raise TypeError(f"hook {self.name!r} takes at most {self.positional_arity} positional argument(s)")
        positional = dict(zip(self.params[: self.positional_arity], 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 collect_with_plugins(self, *args: Any, **kwargs: Any) -> list[PluginResult[Any]]:
        """Collect non-None results together with their registered plugin names."""
        if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
            raise TypeError(f"hook {self.name!r} is not a collecting hook")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return cast(list[PluginResult[Any]], self._execute(kwargs, self._nonwrappers, with_plugins=True))

    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)
        # Dispatch to the current impls and fire their callbacks first: if either the
        # dispatch or a callback raises, the call fails without recording the event, so a
        # later plugin never replays (and re-fails on) a call that did not fully succeed.
        outcomes = self._collect(kwargs)
        for outcome in outcomes:
            if result_callback is not None:
                result_callback(outcome)
        # Record only after a fully successful call, snapshotting the event so a later
        # mutation of the caller's dict can't change what plugins replay.
        self._history.append((dict(kwargs), result_callback))

    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], *, with_plugins: bool = False) -> 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_with_plugins(kwargs, nonwrappers) if with_plugins else 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")
                try:
                    next(generator)  # advance to the yield
                except StopIteration:
                    raise RuntimeError(f"wrapper for {self.name!r} must yield exactly once") from None
                started.append(generator)
            result = self._core_with_plugins(kwargs, nonwrappers) if with_plugins else 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

    @staticmethod
    def _core_with_plugins(kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> list[PluginResult[Any]]:
        """Collect non-None results with provenance in ordinary dispatch order."""
        results: list[PluginResult[Any]] = []
        for impl in nonwrappers:
            outcome = impl.call(kwargs)
            if outcome is not None:
                results.append(PluginResult(impl.plugin_name, 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.
                # Closing the offender can itself raise (a failing `finally`); that error
                # then carries the contract violation as its cause and unwinding continues.
                exc = RuntimeError(f"wrapper for {self.name!r} must yield exactly once")
                try:
                    generator.close()
                except BaseException as close_exc:  # noqa: BLE001 - keep unwinding the rest
                    close_exc.__cause__ = exc
                    exc = close_exc
        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 
154
155
156
157
158
159
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)
    if self.positional_arity < 0:
        self.positional_arity = len(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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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 
183
184
185
186
187
188
189
190
191
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 
193
194
195
196
197
198
199
200
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 
202
203
204
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 
237
238
239
240
241
242
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)
collect_with_plugins(*args, **kwargs)

Collect non-None results together with their registered plugin names.

Source code in src/pluginkit/manager.py 
244
245
246
247
248
249
def collect_with_plugins(self, *args: Any, **kwargs: Any) -> list[PluginResult[Any]]:
    """Collect non-None results together with their registered plugin names."""
    if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
        raise TypeError(f"hook {self.name!r} is not a collecting hook")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return cast(list[PluginResult[Any]], self._execute(kwargs, self._nonwrappers, with_plugins=True))
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 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
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 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
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)
    # Dispatch to the current impls and fire their callbacks first: if either the
    # dispatch or a callback raises, the call fails without recording the event, so a
    # later plugin never replays (and re-fails on) a call that did not fully succeed.
    outcomes = self._collect(kwargs)
    for outcome in outcomes:
        if result_callback is not None:
            result_callback(outcome)
    # Record only after a fully successful call, snapshotting the event so a later
    # mutation of the caller's dict can't change what plugins replay.
    self._history.append((dict(kwargs), result_callback))
implementations()

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

Source code in src/pluginkit/manager.py 
400
401
402
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 
404
405
406
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 
413
414
415
416
417
418
419
420
421
422
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

    def collect_with_plugins(self, *args: P.args, **kwargs: P.kwargs) -> list[PluginResult[R]]:
        """Collect results paired with the plugin names that produced them."""
        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 
416
417
418
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
collect_with_plugins(*args, **kwargs)

Collect results paired with the plugin names that produced them.

Source code in src/pluginkit/manager.py 
420
421
422
def collect_with_plugins(self, *args: P.args, **kwargs: P.kwargs) -> list[PluginResult[R]]:
    """Collect results paired with the plugin names that produced them."""
    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 
425
426
427
428
429
430
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 
428
429
430
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 
433
434
435
436
437
438
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 
436
437
438
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 
441
442
443
444
445
446
447
448
449
450
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 
448
449
450
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 
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
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 
456
457
458
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 
472
473
474
475
476
477
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 
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
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
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._spec_callers: dict[int, tuple[object, HookCaller]] = {}
        self._lock = threading.RLock()

    def add_extension_points(self, namespace: object) -> None:
        """Scan a module (or object) for extension points and create callers.

        Every caller is built and validated before any is registered, so a namespace
        with an invalid spec leaves the manager unchanged rather than half-configured.
        """
        with self._lock:
            new_callers: list[HookCaller] = []
            new_specs: list[tuple[object, HookCaller]] = []
            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)
                arity = _positional_arity(signature, "extension point", member_name)
                params = tuple(signature.parameters)
                defaults = {
                    name: parameter.default
                    for name, parameter in signature.parameters.items()
                    if parameter.default is not inspect.Parameter.empty
                }
                if self.hook._get_caller(member_name) is not None:
                    raise ValueError(
                        f"extension point {member_name!r} is already registered; "
                        f"re-adding it would drop the implementations already wired to it"
                    )
                self._validate_spec(member_name, spec, signature)
                caller = self._make_caller(member_name, spec, params, defaults, arity)
                new_callers.append(caller)
                new_specs.append((self._spec_identity(member), caller))
            for caller in new_callers:
                self.hook._add_caller(caller)
            for member, caller in new_specs:
                self._spec_callers[id(member)] = (member, caller)

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

    @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)."""
        identity = self._spec_identity(spec)
        name = getattr(spec, "__name__", None)
        if not isinstance(name, str):
            raise TypeError("caller() expects an @extension_point-decorated function")
        marker = getattr(spec, self._spec_attribute, None)
        if not isinstance(marker, ExtensionPointOpts):
            raise TypeError(f"caller() expects an extension point decorated for project {self.project_name!r}")
        registered = self._spec_callers.get(id(identity))
        if registered is None or registered[0] is not identity:
            raise PluginValidationError(
                self.project_name, f"unknown extension point {name!r}; call add_extension_points() first"
            )
        return registered[1]

    @staticmethod
    def _spec_identity(spec: object) -> object:
        """Normalize bound methods to their stable underlying function identity."""
        return getattr(spec, "__func__", spec)

    @staticmethod
    def _validate_spec(name: str, spec: ExtensionPointOpts, signature: inspect.Signature) -> None:
        """Reject contradictory or impossible spec option combinations."""
        params = tuple(signature.parameters)
        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")
        if spec.pipeline:
            threaded = signature.parameters[params[0]].annotation
            returned = signature.return_annotation
            if _is_concrete_annotation(threaded) and _is_concrete_annotation(returned) and threaded != returned:
                raise ValueError(
                    f"pipeline hook {name!r} must return its threaded first argument type; "
                    f"got {threaded!r} -> {returned!r}"
                )

    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:
                    if self._name2plugin.get(plugin_name) is not plugin:
                        # A historic replay unregistered or blocked the plugin from inside
                        # one of its own implementations. Stop wiring the rest so no hook
                        # outlives the registration it belonged to.
                        for hook_caller in self.hook._all_callers():
                            hook_caller.remove_plugin(plugin_name)
                        return plugin_name
                    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 load_entrypoints_report(self, group: str) -> EntryPointLoadReport:
        """Discover entry points resiliently and return structured outcomes.

        Unlike strict :meth:`load_entrypoints`, this method always continues after
        individual failures. The original exception is retained for host reporting.
        """
        loaded: list[str] = []
        already_loaded: list[str] = []
        blocked: list[str] = []
        failed: list[EntryPointFailure] = []
        for entry_point in entry_points(group=group):
            if entry_point.name in self._name2plugin:
                already_loaded.append(entry_point.name)
                continue
            if entry_point.name in self._blocked:
                blocked.append(entry_point.name)
                continue
            try:
                plugin = entry_point.load()
                self.register(plugin, name=entry_point.name)
            except Exception as error:
                failed.append(EntryPointFailure(entry_point.name, error))
            else:
                loaded.append(entry_point.name)
        return EntryPointLoadReport(
            loaded=tuple(loaded),
            already_loaded=tuple(already_loaded),
            blocked=tuple(blocked),
            failed=tuple(failed),
        )

    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")
            if opts.tryfirst and opts.trylast:
                raise PluginValidationError(
                    plugin_name, f"hook {hook_name!r} cannot combine tryfirst=True and trylast=True"
                )
            try:
                impl = HookImpl.from_function(plugin_name, member, opts)
            except ValueError as exc:
                raise PluginValidationError(plugin_name, str(exc)) from exc
            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 
483
484
485
486
487
488
489
490
491
492
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._spec_callers: dict[int, tuple[object, HookCaller]] = {}
    self._lock = threading.RLock()
add_extension_points(namespace)

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

Every caller is built and validated before any is registered, so a namespace with an invalid spec leaves the manager unchanged rather than half-configured.

Source code in src/pluginkit/manager.py 
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
def add_extension_points(self, namespace: object) -> None:
    """Scan a module (or object) for extension points and create callers.

    Every caller is built and validated before any is registered, so a namespace
    with an invalid spec leaves the manager unchanged rather than half-configured.
    """
    with self._lock:
        new_callers: list[HookCaller] = []
        new_specs: list[tuple[object, HookCaller]] = []
        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)
            arity = _positional_arity(signature, "extension point", member_name)
            params = tuple(signature.parameters)
            defaults = {
                name: parameter.default
                for name, parameter in signature.parameters.items()
                if parameter.default is not inspect.Parameter.empty
            }
            if self.hook._get_caller(member_name) is not None:
                raise ValueError(
                    f"extension point {member_name!r} is already registered; "
                    f"re-adding it would drop the implementations already wired to it"
                )
            self._validate_spec(member_name, spec, signature)
            caller = self._make_caller(member_name, spec, params, defaults, arity)
            new_callers.append(caller)
            new_specs.append((self._spec_identity(member), caller))
        for caller in new_callers:
            self.hook._add_caller(caller)
        for member, caller in new_specs:
            self._spec_callers[id(member)] = (member, caller)
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 
544
545
546
547
548
549
550
551
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 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
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:
                if self._name2plugin.get(plugin_name) is not plugin:
                    # A historic replay unregistered or blocked the plugin from inside
                    # one of its own implementations. Stop wiring the rest so no hook
                    # outlives the registration it belonged to.
                    for hook_caller in self.hook._all_callers():
                        hook_caller.remove_plugin(plugin_name)
                    return plugin_name
                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 
632
633
634
635
636
637
638
639
640
641
642
643
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 
645
646
647
648
649
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 
651
652
653
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 
655
656
657
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 
659
660
661
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 
663
664
665
666
667
668
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 
670
671
672
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 
674
675
676
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 
678
679
680
681
682
683
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 
685
686
687
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 
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
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
load_entrypoints_report(group)

Discover entry points resiliently and return structured outcomes.

Unlike strict :meth:load_entrypoints, this method always continues after individual failures. The original exception is retained for host reporting.

Source code in src/pluginkit/manager.py 
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
def load_entrypoints_report(self, group: str) -> EntryPointLoadReport:
    """Discover entry points resiliently and return structured outcomes.

    Unlike strict :meth:`load_entrypoints`, this method always continues after
    individual failures. The original exception is retained for host reporting.
    """
    loaded: list[str] = []
    already_loaded: list[str] = []
    blocked: list[str] = []
    failed: list[EntryPointFailure] = []
    for entry_point in entry_points(group=group):
        if entry_point.name in self._name2plugin:
            already_loaded.append(entry_point.name)
            continue
        if entry_point.name in self._blocked:
            blocked.append(entry_point.name)
            continue
        try:
            plugin = entry_point.load()
            self.register(plugin, name=entry_point.name)
        except Exception as error:
            failed.append(EntryPointFailure(entry_point.name, error))
        else:
            loaded.append(entry_point.name)
    return EntryPointLoadReport(
        loaded=tuple(loaded),
        already_loaded=tuple(already_loaded),
        blocked=tuple(blocked),
        failed=tuple(failed),
    )

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
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
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 collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
        self, *args: Any, **kwargs: Any
    ) -> list[PluginResult[Any]]:
        """Await collecting implementations and preserve each result's plugin name."""
        if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
            raise TypeError(f"hook {self.name!r} is not a collecting hook")
        kwargs = self.check_arguments(self._bind(args, kwargs))
        return cast(list[PluginResult[Any]], await self._execute_async(kwargs, self._nonwrappers, with_plugins=True))

    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], *, with_plugins: bool = False
    ) -> 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")
                try:
                    await generator.__anext__()  # advance to the yield
                except StopAsyncIteration:
                    raise RuntimeError(f"async wrapper for {self.name!r} must yield exactly once") from None
                started.append(generator)
            result = (
                await self._core_with_plugins_async(kwargs, nonwrappers)
                if with_plugins
                else 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

    @staticmethod
    async def _core_with_plugins_async(kwargs: dict[str, Any], nonwrappers: list[HookImpl]) -> list[PluginResult[Any]]:
        """Await and collect non-None results with provenance."""
        results: list[PluginResult[Any]] = []
        for impl in nonwrappers:
            outcome = await _maybe_await(impl.call(kwargs))
            if outcome is not None:
                results.append(PluginResult(impl.plugin_name, outcome))
        return 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. A failing
                # aclose() carries the contract violation as its cause and does not stop
                # the unwind either.
                exc = RuntimeError(f"async wrapper for {self.name!r} must yield exactly once")
                try:
                    await generator.aclose()
                except BaseException as close_exc:  # noqa: BLE001 - keep unwinding the rest
                    close_exc.__cause__ = exc
                    exc = close_exc
        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)
collect_with_plugins(*args, **kwargs) async

Await collecting implementations and preserve each result's plugin name.

Source code in src/pluginkit/aio.py 
39
40
41
42
43
44
45
46
async def collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
    self, *args: Any, **kwargs: Any
) -> list[PluginResult[Any]]:
    """Await collecting implementations and preserve each result's plugin name."""
    if self.spec.firstresult or self.spec.pipeline or self.spec.historic:
        raise TypeError(f"hook {self.name!r} is not a collecting hook")
    kwargs = self.check_arguments(self._bind(args, kwargs))
    return cast(list[PluginResult[Any]], await self._execute_async(kwargs, self._nonwrappers, with_plugins=True))
call_extra(functions, kwargs) async

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

Source code in src/pluginkit/aio.py 
48
49
50
51
52
53
54
55
56
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 
58
59
60
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 
154
155
156
157
158
159
160
161
162
163
164
165
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

    async def collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
        self, *args: P.args, **kwargs: P.kwargs
    ) -> list[PluginResult[R]]:
        """Await results paired with the plugin names that produced them."""
        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 
157
158
159
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
collect_with_plugins(*args, **kwargs) async

Await results paired with the plugin names that produced them.

Source code in src/pluginkit/aio.py 
161
162
163
164
165
async def collect_with_plugins(  # type: ignore[override]  # async caller returns an awaitable result
    self, *args: P.args, **kwargs: P.kwargs
) -> list[PluginResult[R]]:
    """Await results paired with the plugin names that produced them."""
    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 
168
169
170
171
172
173
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 
171
172
173
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 
176
177
178
179
180
181
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 
179
180
181
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 
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
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], arity: int
    ) -> HookCaller:
        """Build an AsyncHookCaller; historic extension points are not supported async-side."""
        if spec.historic:
            raise ValueError(
                f"historic extension point {name!r} is not supported by AsyncPluginManager "
                f"(historic replay requires synchronous dispatch)"
            )
        return AsyncHookCaller(name=name, spec=spec, params=params, defaults=defaults, positional_arity=arity)

    @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 
204
205
206
207
208
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."""