onsrap.pipeline module

class onsrap.pipeline.Pipeline(name: str | None = None, backend: str = 'python', config: PipelineConfig | Mapping[str, Any] | str | Path | None = None, stages: Sequence[Stage | Mapping[str, Any] | str | Path | Callable[[...], Any]] | None = None, dependencies: Mapping[str, Sequence[str]] | None = None, logger: Logger | None = None, executor: StageExecutor | PythonStageExecutor | None = None)

Bases: object

Represents an end-to-end code run. This class brings together class instances from other modules within the package to establish what the Pipeline is.

Sets up the metadata, configurations, logging, and executors required to run the Pipeline. Assigns multiple attributes including those not initialised such as, id, graph, manifest, and last_run. These take the forms of other classes defined in other modules within this package.

Parameters:

  • name (str or None) – What the pipeline is called.

  • backend (str, default = "python") – The system used to run the pipeline.

  • config (PipelineConfig | Mapping[str, Any] | str | Path | None) – The instance containing the required information on running the Pipeline.

  • stages (sequence of Stage, Mapping[str, Any], str, Path, Callable, or None.) – The required steps within the Pipeline.

  • logger (Logger or None) – The system that is used to track the progress of the Pipeline.

  • executor (StageExecutor or None) – The way that the Pipeline is actively run.

add_dependencies(*dependencies: Mapping[str, Sequence[str]]) → None

Add dependency mappings to stages already registered on the Pipeline.

Each positional argument must be a mapping whose keys identify target stages by stage name, source-path filename, full source path, or callable name. Values are normalized, appended to the matching stage’s existing dependencies, de-duplicated in first-seen order, and merged into Pipeline.dependencies before the execution graph is rebuilt.

Parameters:

*dependencies (Mapping[str, Sequence[str]]) – One or more dependency mappings to merge into the Pipeline.

Raises:

PipelineInitialisationError – If a dependency payload is not provided as a mapping.

add_stage(*stages: Stage | Mapping[str, Any] | str | Path | Callable[[...], Any], stage_configs: StageConfig | Mapping[str, Any] | str | Path | Iterable[StageConfig | Mapping[str, Any] | str | Path] | None = None, enable_stages: bool = False) → None

Adds one or more steps to the Pipeline.

enable_stagesbool, default False

Whether to enable the added stages immediately. Default is False and is recommended.

Creates a list called added_stages that runs the _coerce_stage() method to extract the information from the given stages parameter. It then appends this list to the stages attribute of the Pipeline class, adds any stage configuration that was provided alongside those stages, and updates the StageGraph using the _rebuild_graph() method.

Parameters:

  • stages (Stage | Mapping[str, Any] | str | Path | Callable[..., Any]) – The new steps being added to the Pipeline.

  • stage_configs (StageConfig | Mapping[str, Any] | str | Path | Iterable[StageConfig | Mapping[str, Any] | str | Path] | None) – Optional stage configuration payloads to add alongside the stages.

add_stage_config(stage_config: StageConfig | Mapping[str, Any] | str | Path, *, name: str | None = None) → None

Add or replace a StageConfig attached to the Pipeline.

Parameters:

  • stage_config (StageConfig | Mapping[str, Any] | str | Path) – The stage configuration information to add to the Pipeline.

  • name (str or None, keyword-only) – Optional stage name used when the parsed configuration payload does not identify the stage on its own.

create_stage_config(stage_config: StageConfig | Mapping[str, Any] | str | Path, *, name: str | None = None) → StageConfig

Normalize a stage-configuration payload into a StageConfig instance.

This compatibility helper accepts direct stage payloads, stage-name keyed mappings, and composite configuration payloads or files containing a stage_configuration section.

disable_stage(*stage_name: str | list[str]) → None

Mark one or more stages as disabled in the run selection.

When in implicit “run all” mode (stages_to_run is empty), calling disable_stage switches the pipeline into explicit stage-selection mode: every currently registered stage is first marked enabled, then the requested stages are set to False. The execution graph is rebuilt after the change.

Parameters:

stage_name (str or list[str]) – One or more stage names to disable.

enable_stage(*stage_name: str | list[str]) → None

Mark one or more stages as enabled in the run selection.

In implicit “run all” mode (stages_to_run is empty), this is a no-op because every registered stage already participates in the execution graph. In explicit mode the requested stages are marked True in stages_to_run and the execution graph is rebuilt to reflect the change.

Parameters:

stage_name (str or list[str]) – One or more stage names to enable.

classmethod from_config(config: Mapping[str, Any] | str | Path, name: str | None = None, backend: str = 'python', logger: Logger | None = None, executor: StageExecutor | None = None) → Pipeline

Construct a pipeline directly from a composite configuration payload or file.

This is the preferred entrypoint when configuration defines both pipeline-level settings and the stage-level configuration that should be injected at runtime.

classmethod from_dict(config: PipelineConfig | Mapping[str, Any] | str | Path, name: str | None = None, backend: str = 'python', logger: Logger | None = None, executor: StageExecutor | None = None) → Pipeline

Extracts information from a dictionary to configure a Pipeline instance as well as what the Pipeline runs.

Parameters:

config (PipelineConfig | Mapping[str, Any] | str | Path) – The object containing the information needed to run the Pipeline.

Return type:

A Pipeline class instance.

classmethod from_files(file_paths: Iterable[str | Path], *, name: str | None = None, backend: str = 'python', config: PipelineConfig | Mapping[str, Any] | str | Path | None = None, dependencies: Mapping[str, Sequence[str]] | None = None, logger: Logger | None = None, executor: StageExecutor | None = None) → Pipeline

Extracts the information from files regarding exactly what is being run in the pipeline and allows for configuration of how the Pipeline is run.

Parameters:

  • file_paths (Iterable[str or Path]) – The files that contain the code for each stage in the pipeline. These are what the Pipeline will run.

  • name (str) – The name of the pipeline.

  • backend (str, default = "python") – The system that the pipeline is written in.

  • config (PipelineConfig | Mapping[str, Any] | str | Path | None) – The high level information required to run this specific pipeline.

  • dependencies (Mapping[str, Sequence[str]] or None) – An object containing which stages are required to be run before other stages.

  • logger (Logger class or None) – The logging sysem used for this Pipeline run.

  • executor (StageExecutor class or None) – The information on exactly how to run the Pipeline.

Return type:

A Pipeline class instance.

ordered_stages() → list[Stage]

Return the effective stages in dependency-respecting execution order.

This is the primary method used by PipelineRunner to determine what to execute. Only stages that are part of the current execution graph appear here; stages disabled via PipelineConfig.stages_to_run are absent even if they are registered in Pipeline.stages.

run() → PipelineRun

Returns an instance of PipelineRunner which actually runs the pipeline.

validate() → Pipeline

Confirm that the pipeline is ready to run.

Validates source files for every stage in the current execution graph, checks that all stage-configuration names correspond to a known stage, and validates the execution graph for structural consistency. Disabled stages are excluded from source-file validation because they will not be executed.