2.27.2.4. pycsamt.pipeline._plugins#

Entry-point discovery for third-party pipeline steps.

A plugin package registers pipeline steps by declaring a zero-argument callable under the pycsamt.pipeline.steps entry-point group, e.g. in its pyproject.toml:

[project.entry-points."pycsamt.pipeline.steps"]
my_plugin = "my_package.pipeline_steps:register"

where my_package.pipeline_steps.register() calls pycsamt.pipeline.register_step() for whatever it contributes.

Discovery is never run automatically on import pycsamt.pipeline — no third-party code should execute just because a package happens to be installed. Call discover_plugins() explicitly, or use the pycsamt pipe CLI, which calls it for you before every command.

Functions

discover_plugins(*[, group, on_error])

Load every pycsamt.pipeline.steps entry point and run it.

Classes

PluginLoadResult(name, ok[, error])

Outcome of loading one pycsamt.pipeline.steps entry point.
class pycsamt.pipeline._plugins.PluginLoadResult(name, ok, error=None)[source]

Bases: object

Outcome of loading one pycsamt.pipeline.steps entry point.

Variables:

  • name (str) – The entry point’s registered name (left-hand side in [project.entry-points."pycsamt.pipeline.steps"]).

  • ok (bool) – Whether the entry point loaded and ran without raising.

  • error (str | None) – str(exception) when ok is False, else None.

Parameters:
name: str
ok: bool
error: str | None = None
pycsamt.pipeline._plugins.discover_plugins(*, group='pycsamt.pipeline.steps', on_error='warn')[source]

Load every pycsamt.pipeline.steps entry point and run it.

Each entry point must resolve to a zero-argument callable; it is expected to call pycsamt.pipeline.register_step() itself for whatever steps it contributes. Never called automatically — see the module docstring.

Parameters:

  • group (str) – Entry-point group to scan. Defaults to ENTRY_POINT_GROUP.

  • on_error (str) – "warn" (default): a plugin that fails to load or raises is reported as a failed PluginLoadResult and a UserWarning; discovery continues with the remaining plugins. "raise": the first failure propagates immediately.

Returns:

One entry per discovered entry point, in discovery order.

Return type:

list[PluginLoadResult]