onsrap package

class onsrap.Catalog(name: 'str', description: 'str', contents: 'dict[str, Any]')

Bases: object

contents: dict[str, Any]
description: str
name: str
exception onsrap.DependencyCycleError

Bases: PipelineValidationError

Raised when the stage graph contains a cycle. Child class with PipelineValidationError as the parent class.

exception onsrap.DuplicateStageError

Bases: PipelineValidationError

Raised when two stages share the same name. Child class with PipelineValidationError as the parent class.

class onsrap.ExecutionContext(pipeline_name: str, run_id: str, config: ~onsrap.models.PipelineConfig, logger: ~onsrap.logger.Logger, run_dir: ~pathlib.Path, started_at: ~datetime.datetime = <factory>, working_directory: ~pathlib.Path = <factory>, stage_results: dict[str, ~onsrap.models.StageResult] = <factory>, stage_configs: dict[str, ~onsrap.models.StageConfig] = <factory>, variables: dict[str, ~typing.Any] = <factory>, active_stage_name: str | None = None, global_config: ~onsrap.models.GlobalConfig | None = None)

Bases: object

Holds information needed to run the pipeline.

Parameters:

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

  • run_id (str) – The unique identifier for the current run of the pipeline.

  • config (PipelineConfig class instance) – The configuration required for the pipeline.

  • logger (Logger class instance) – The logger used for this pipeline run.

  • run_dir (Path) – The directory that the run saved to.

  • started_at (datetime, default = current time) – The time that the pipeline run started.

  • working_directory (Path, default = current working directory) – The directory that the work is taking place in.

  • stage_results (dict[str, StageResult], default = dict) – Stores the logs for the stage run.

  • stage_configs (dict[str, StageConfig], default = dict) – Stage-name keyed configuration mapping resolved by the Pipeline.

  • variables (dict[str, Any], default = dict) – Stores relevant variables regarding the stage run and their results.

  • active_stage_name (str or None, default = None) – Name of the stage currently being executed. Used to expose stage_config.

  • global_config (GlobalConfig or None, default = None) – Variables which are parsed to all stages throughout the pipeline.

active_stage_name: str | None = None
config: PipelineConfig
get_data_dir() → Path

Establishes the filepath that the data is held in.

Returns:

The file path for the location of the data being used in the pipeline.

Return type:

Path

get_stage_config(stage: str | None = None, with_global: bool = True, vars_only: bool = True) → dict[str, Any] | StageConfig | None

Returns the configuration for the stage currently being executed, with optional arguments.

Optional argument vars_only can be set to False to return the full StageConfig instance, rather than just the variables dictionary.

If you want to access metadata or dataframes from the StageConfig, you must set vars_only to False.

Parameters:

  • stage (str) – The name of the stage to get the configuration for.

  • vars_only (bool, default = True) – If True, returns only the variables dictionary from the StageConfig. If False, returns the full StageConfig instance.

Returns:

The parameters contained within the configuration for the currently active stage. If vars_only is set to False, returns the StageConfig object itself, containing all attributes including variables, metadata, and dataframes.

Return type:

dict[str, Any] or StageConfig or None

global_config: GlobalConfig | None = None
logger: Logger
pipeline_name: str
record(result: StageResult) → StageResult

Extracts key information from StageResult.

Saves all information on the results of the Stage to the stage_results attribute and exclusively metadata outputs regarding the run to the variables attribute.

Parameters:

result (StageResult) – An instance of a StageResult class which is created from the Executor classes (StageExecutor, PythonStageExecutor).

Returns:

An unchanged StageResult instance.

Return type:

result

resolve_given_path(stage_name: str | None, path_name: str | None, file_name: str | None, root: Path, add_folder: list[str] | str | None = None) → Path

Returns a file path for a requested item.

This investigates the result of a previous stage to extract a selected path. If the path is not available, it creates a path using a root previously derived in main.py, the chosen directory within the root (optional), and the file path.

Parameters:

  • stage_name (str) – The name of the stage where the path was outputted.

  • path_name (str) – The name for the path within the stage results. This will be the key from the key/value pair within the output of the previous stage.

  • file_name (str) – The name of the file that you are trying to access the Path for.

  • root (Path) – The file path for the root of the directory. This should be denoted through other methods.

  • add_folder (list[str] | str | None, default = None) – Additional folder name/s to add into the returned file path.

Returns:

The file path where data has previously been saved to to allow for extraction of that data throughout the pipeline.

Return type:

Path

resolve_output_root() → Path

Establishes the filepath that the outputs are going to be saved to.

Returns:

The file path for the outputs of the run to be saved to.

Return type:

Path

result_for(stage_name: str) → StageResult | None

Getter function that returns the stage_results for a specific Stage.

Parameters:

stage_name (str) – The name of the Stage that you are calling the results for.

Returns:

Attribute for the specific Stage named.

Return type:

stage_results

run_dir: Path
run_id: str
set_active_stage(stage_name: str | None) → None

Mark the stage currently being executed so stage_config resolves correctly.

property stage_config: StageConfig | None

Return the configuration for the stage currently being executed.

The preferred access method for this is get_stage_config() which allows for optional arguments to return the full StageConfig instance or just the variables dictionary.

This property is None outside an active stage run.

stage_config_for(stage_name: str | None) → StageConfig | None

Return the configuration registered for stage_name.

Unlike stage_config, this helper does not depend on the currently active stage and can be used to inspect any known stage configuration.

Parameters:

stage_name (str or None) – Name of the stage whose configuration should be returned.

stage_configs: dict[str, StageConfig]
property stage_outputs: dict[str, Any]

Creates a stage_outputs attribute for the ExecutionContext class.

Extracts the `outputs attribute from the stage_results class for each Stage name.

Returns:

Dictionary containing the name of the stage and the associated outputs of the run.

Return type:

stage_outputs

stage_results: dict[str, StageResult]
started_at: datetime
variables: dict[str, Any]
working_directory: Path
class onsrap.GlobalConfig(_variables: dict[str, ~typing.Any] = <factory>, exclusion: dict[str, ~typing.Any] = <factory>)

Bases: object

Holds configuration that should be exposed to all stages at runtime.

Parameters:

_variables (dict[str, Any]) – Variables that should be parsed to all stages throughout the pipeline.

exclusion: dict[str, Any]
classmethod from_dict(data: Mapping[str, Any] | None) → GlobalConfig

Build a GlobalConfig from a mapping loaded from code or configuration files.

Parameters:

  • data (Mapping[str, Any] | None) – Raw configuration payload for the global configuration.

  • exclusions (dict[str, Any] or None) – A lookup of which global variables should be excluded from each stage.

Returns:

A global configuration object.

Return type:

GlobalConfig

get_attributes(keep_exclusion: Literal[True] = True) → tuple[dict[str, Any], dict[str, Any]]
get_attributes(keep_exclusion: Literal[False]) → dict[str, Any]

Return a copy of the global variables, optionally excluding any variables specified in the exclusion list.

Parameters:

keep_exclusion (bool, default = True) – If True, return both _variables and exclusion. If False, return only the variables and not the exclusion list.

Returns:

  • ``self._variables`` (dict[str, Any]) – All global variables for the pipeline.

  • ``self.exclusion`` (dict[str, Any]) – The exclusion list of global variables for each stage. Only returned if keep_exclusion is True.

class onsrap.LogConfig(log_dir: str = 'logs/', log_level: str = 'INFO', logger_name: str = 'onsrap')

Bases: object

Data class which holds information regarding how the logs are set up.

Parameters:

  • log_dir (str, default = "logs/") – The directory where all logs are stored for the Pipeline.

  • log_level (str, default = "INFO") – Denotes how severe the log message is.

  • logger_name (str, default = "onsrap") – The name of the logging system.

log_dir: str = 'logs/'
log_level: str = 'INFO'
logger_name: str = 'onsrap'
class onsrap.Logger(log_dir: str | Path = 'logs/', log_level: str = 'INFO')

Bases: object

Creates a logging system.

This system creates a logging directory and enables writing the log messages to both console and the logging files. It allows configurable logging levels to adjust for severity and avoids duplicating logging messages or handlers. If the logger is unable to write to a file, the logging continues using only the console handler.

Parameters:

  • log_dir (str or Path, default = "logs/") – The directory where you’d like your logs stored.

  • log_level (str, default = "INFO") – The severity of the log.

event(message: str, **kwargs: Any) → None

Logs a named event with optional structured context.

Parameters:

  • message (str) – The main description of the event to be logged.

  • **kwargs (Any) – Additional information to be recorded in the log record.

extract_historical_run_ids(run_root: Path, name: str) → list[dict[str, Any]]

Extracts historical run IDs from the log files.

Parameters:

  • run_root (Path) – The root directory where the historical runs are stored.

  • name (str) – The name of the pipeline for which to extract historical run IDs.

Returns:

A list of dictionaries containing run_id, timestamp, and run_dir for each historical run.

Return type:

list[dict[str, Any]]

warning(message: str, **kwargs: Any) → None

Logs a warning message with optional structured context.

Parameters:

  • message (str) – The main description of the warning to be logged.

  • **kwargs (Any) – Additional information to be recorded in the log record.

exception onsrap.MissingDependencyError

Bases: PipelineValidationError

Raised when a stage depends on an unknown stage. Child class with PipelineValidationError as the parent class.

exception onsrap.OnsrapError

Bases: Exception

Base exception for onsrap.

class onsrap.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.

class onsrap.PipelineConfig(name: str | None = None, stages_to_run: dict[str, bool] | None = None, backend: str = 'python', work_dir: ~pathlib.Path = <factory>, project_root: ~pathlib.Path | None = None, output_dir: ~pathlib.Path | None = None, log_dir: ~pathlib.Path = <factory>, data_dir: ~pathlib.Path = <factory>, allow_subprocess_fallback: bool = True, python_executable: str | None = None, metadata: dict[str, ~typing.Any] = <factory>, overwrite: bool = False)

Bases: object

Holds information required to run the whole pipeline.

Parameters:

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

  • stages_to_run (dict[str, bool], optional) – A dictionary of all stage names alongside a boolean value that indicates whether the stage should be run or not.

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

  • work_dir (Path) – The directory to run the Pipeline in.

  • project_root (Path) – The top level directory for the whole project.

  • log_dir (Path) – The directory to store the logs in.

  • data_dir (Path) – The directory where the data is stored.

  • output_dir (Path, optional) – The directory where pipeline outputs should be written. Not used internally by the runner; exposed for stage code to read via context.config.output_dir.

  • allow_subprocess_fallback (bool) – Indicates whether the subprocess system (running the whole file rather than an entrypoint function) should be allowed.

  • python_executable (str, optional) – The name of the executable function for the entrypoint of the pipeline.

  • metadata (dict[str, Any]) – Any additional information on the pipeline.

  • overwrite (bool, default = False) – Indicates whether the pipeline should overwrite previous outputs.

allow_subprocess_fallback: bool = True
backend: str = 'python'
data_dir: Path
classmethod from_any(value: PipelineConfig | Mapping[str, Any] | str | Path | None) → PipelineConfig

Converts one of several datatypes into a PipelineConfig class instance.

Parameters:

value (PipelineConfig, Mapping[str, Any], str, Path, or None) – The object holding metadata on how the Pipeline should run to be converted into a PipelineConfig class instance.

Raises:

TypeError – If the datatype for the object holding information on how the pipeline is run is not a datatype that can be converted to a PipelineConfig.

classmethod from_file(path: Path) → PipelineConfig

Extracts a mapping item from a file containing information about how the pipeline should run.

Then calls the from_mapping() method to extract the information.

Parameters:

path (Path) – The file path containing information to be converted into a PipelineConfig instance.

Return type:

PipelineConfig class instance.

Raises:

  • FileNotFoundError – If the file path does not exist.

  • TypeError – If the file containing information about how the Pipeline runs does not contain a mapping type.

classmethod from_mapping(data: Mapping[str, Any]) → PipelineConfig

Extracts information from a mapping datatype and returns a PipelineConfig instance.

Parameters:

data (Mapping[str, Any]) – The information to be converted into a PipelineConfig instance.

Return type:

PipelineConfig class instance

log_dir: Path
metadata: dict[str, Any]
name: str | None = None
output_dir: Path | None = None
overwrite: bool = False
project_root: Path | None = None
python_executable: str | None = None
stages_to_run: dict[str, bool] | None = None
to_dict() → dict[str, Any]

Returns a prescriptive expression of the attributes within the PipelineConfig instance that allows for easier processing by the user.

work_dir: Path
class onsrap.PipelineRun(manifest: ~onsrap.models.RunManifest, status: ~onsrap.models.PipelineStatus, started_at: ~datetime.datetime, completed_at: ~datetime.datetime, stage_results: list[~onsrap.models.StageResult] = <factory>, stage_outputs: dict[str, ~typing.Any] = <factory>)

Bases: object

Holds information about how the whole Pipeline ran.

Parameters:

  • manifest (RunManifest class instance) – Metadata on how the specific run has gone.

  • status (PipelineStatus class instance) – Whether the Pipeline ran successfully or if there were errors.

  • started_at (datetime) – The date and time the Pipeline started.

  • completed_at (datetime) – The date and time the Pipeline ended.

  • stage_results (list[StageResult]) – Holds the results for every stage run as part of the Pipeline.

  • stage_outputs (dict[str, Any]) – Holds the outputs from all stages run as part of the Pipeline.

completed_at: datetime
classmethod load_pipeline_run_for_historical_run(file_path: Path) → PipelineRun

Load a previously executed pipeline run from a YAML file.

This function is used to load the state of a pipeline run that has been saved to a YAML file. It reads the file, parses the YAML content, and reconstructs the PipelineRun object.

Parameters:

file_path (Path) – The path to the YAML file containing the saved pipeline run.

Returns:

The reconstructed PipelineRun object.

Return type:

PipelineRun

Raises:

FileNotFoundError – If the specified file does not exist.

manifest: RunManifest
result_for(stage_name: str) → StageResult | None

Extracts the results for a specific stage.

Parameters:

stage_name (str) – The name of the Stage that you are requesting the results for.

stage_outputs: dict[str, Any]
stage_results: list[StageResult]
started_at: datetime
status: PipelineStatus
property succeeded: bool

Creates a new attribute in the PipelineRun class called succeeded that contains a boolean value indicating if the Pipeline was a success or not. Updates the status attribute to record that the Pipeline ran successfully.

class onsrap.PipelineRunner(logger: Logger | None = None)

Bases: object

Represents the information required to run the Pipeline.

Parameters:

logger (Logger class type) – Information used to log progress throughout the Pipeline.

run(pipeline: Pipeline) → PipelineRun

Method that runs a Pipeline instance.

This method validates the source information, establishes the directories and the context to run the pipeline within, sets out the manifest for the run, attempts to run the stages in the order outlined by the StageGraph instance and logs all progress alongside relevant statuses. Before each stage executes, the runner binds the current stage name onto the ExecutionContext so context.stage_config resolves to the correct stage-specific configuration.

It returns a PipelineRun instance containing metadata and logging information for the specific run of the whole Pipeline.

Parameters:

pipeline (Pipeline) – A Pipeline instance that this method will run.

Raises:

StageExecutionError – If the stage is unable to be run. Logs will be created to show a failed stage.

class onsrap.PipelineStatus(value)

Bases: str, Enum

Class to hold information on how the Pipeline has run.

FAILED = 'failed'
PENDING = 'pending'
RUNNING = 'running'
SUCCEEDED = 'succeeded'
exception onsrap.PipelineValidationError

Bases: OnsrapError

Raised when the pipeline definition is invalid. Child class with OnsrapError as the parent class.

class onsrap.PythonStageExecutor(preferred_entrypoints: tuple[str, ...] = ('run', 'main', 'execute'))

Bases: object

Class to run Python Stage.

Contains methods that allow automatic running of individual Stage processes for a pipeline.

execute(stage: Stage, context: ExecutionContext) → StageResult

Main function to select how Stage is run.

Identifies the type of source within the Stage and runs the relevant function for that type.

Parameters:

  • stage (Stage class) – The Stage that is attempting to be run.

  • context (ExecutionContext class) – The metadata required to run the Stage.

Return type:

StageResult instance.

Raises:

StageExecutionError – If the source is not a Path or a callable object.

class onsrap.RAPDataset

Bases: object

class onsrap.RunManifest(rap_name: str = '', run_id: str = '', git_commit: str | None = None, stages_run: list[str] = <factory>, parameters: dict[str, ~typing.Any] = <factory>, inputs: dict[str, ~typing.Any] = <factory>, outputs: dict[str, ~typing.Any] = <factory>, backend: str = 'python', package_versions: list[str] | str = <factory>, timestamp: str = '', reason: str | None = None, user: str | None = None, config: dict[str, ~typing.Any] | None = None)

Bases: object

Holds metadata information about the run.

Parameters:

  • rap_name (str, default = "") – The name of the Pipeline.

  • run_id (str, default = "") – The unique ID of the run.

  • git_commit (str, default = None) – The git commit number for the run, indicating the exact state of the code.

  • stages_run (list[str]) – List of the names of stages that were included in this run.

  • parameters (dict[str, Any])

  • inputs (dict[str, Any])

  • outputs (dict[str, Any])

  • backend (str, default = "python") – The system that the Pipeline will run in.

  • package_versions (list[str] or str) – The package versions that are used in this run.

  • timestamp (str, default = "") – The time that this run started.

  • reason (str, optional, default = None) – The reason that this run took place.

  • user (str, optional, default = None) – The person running this specific run.

backend: str = 'python'
config: dict[str, Any] | None = None
git_commit: str | None = None
inputs: dict[str, Any]
outputs: dict[str, Any]
package_versions: list[str] | str
parameters: dict[str, Any]
rap_name: str = ''
reason: str | None = None
run_id: str = ''
stages_run: list[str]
timestamp: str = ''
user: str | None = None
class onsrap.RuntimeID(id: str, timestamp: datetime, hash: str, short_hash: str)

Bases: object

Holds information regarding individual runs.

Parameters:

  • id (str) – The id number for the run.

  • timestamp (datetime) – The time that the run started.

  • hash (str) – A hashed identifier created with the combined ID and timestamp to create a unique identifier for the run.

  • short_hash (str) – A shortened version of the hash attribute to be used in file names for the runs.

get_hash() → str

Getter function to extract the hash attribute.

get_id() → str

Getter function to extract the id attribute.

get_short_hash() → str

Getter function to extract the short_hash attribute.

get_timestamp() → datetime

Getter function to extract the timestamp attribute.

hash: str
id: str
short_hash: str
timestamp: datetime
class onsrap.Stage(name: str, source: ~pathlib.Path | ~typing.Callable[[...], ~typing.Any] | None = None, dependencies: tuple[str, ...] = <factory>, metadata: dict[str, ~typing.Any] = <factory>, entrypoint: str | None = None, backend: str = 'python')

Bases: object

Represents a single unit of work within a pipeline.

Can be defined by a data source process or a Python script/callable item. Stages may be dependant on other stages and can hold metadata for themselves.

Parameters:

  • name (str) – The name of the Stage being run.

  • source (Path, Callable, or None) – Item being implemented in this Stage. E.g. a file path to a Python script or a function being executed directly. The full file path is gathered if a path is used.

  • dependencies (tuple of strings) – Names of stages that must be completed before this stage is attempted. These are cleaned post initialisation to remove leading/trailing whitespace.

  • metadata (dictionary with string:Any key/value pairs) – Location to store any summary information about the stage being run.

  • entrypoint (str, optional) – Name of the starting script to the pipeline.

  • backend (str, default = "python") – The name of the system that the code runs on.

Raises:

StageConfigurationError – If the stage name is empty or if the source is not a supported type.

backend: str = 'python'
dependencies: tuple[str, ...]
entrypoint: str | None = None
classmethod from_callable(callable_object: Callable[[...], Any], *, name: str | None = None, dependencies: Iterable[str] | str | None = None, metadata: Mapping[str, Any] | None = None, backend: str = 'python') → Stage

Class method that retrieves the name of the Stage from a Callable item.

Parameters:

  • callable_object (Callable with any number of arguments of any type) – The name or file path for the script that the Stage will be running.

  • name (str) – The name of the Stage

  • dependencies (Iterable[str], str, or None) – The Stage/s that need to be complete before the Stage currently attempted.

  • metadata (Mapping[str, Any], or None) – Any supporting information for the Stage being run.

  • entrypoint (str or None) – The name of the first script for the Stage.

  • backend (str, default = "python") – The system that the stage is run on.

Returns:

Stage class instance with collected Stage name, normalised dependencies and metadata, and defined the source as the callable_object.

Return type:

Stage

classmethod from_dict(data: Mapping[str, Any]) → Stage

Class method that converts a dictionary stage into a Stage class instance.

Extracts the values from the key/value pairs in the stage and holds them as attributes.

Parameters:

data (any number of key/value pairs of strings) – The information to convert into a Stage class.

Raises:

StageConfigurationError – If the source is not a suitable type (callable or Path).

Returns:

Stage class instance with collected Stage attributes based on the type of source provided.

Return type:

Stage

classmethod from_file(file_path: str | Path, *, name: str | None = None, dependencies: Iterable[str] | str | None = None, metadata: Mapping[str, Any] | None = None, entrypoint: str | None = None, backend: str = 'python') → Stage

Class method that checks and cleans the file path for the Stage.

Expands file path to its full name and checks whether it exists. The method also cleans other parameters in the Stage class in the return line.

Parameters:

  • file_path (str or Path) – The name or file path for the script that the Stage will be running.

  • name (str) – The name of the Stage

  • dependencies (Iterable[str], str, or None) – The Stage/s that need to be complete before the Stage currently attempted.

  • metadata (Mapping[str, Any], or None) – Any supporting information for the Stage being run.

  • entrypoint (str or None) – The name of the first script for the Stage.

  • backend (str, default = "python") – The system that the Stage is run on.

Raises:

StageConfigurationError – If the file path does not exist

Returns:

Stage class instance with cleaned/checked file path, dependencies, and metadata

Return type:

Stage

metadata: dict[str, Any]
name: str
run(context: ExecutionContext, executor: StageExecutor) → StageResult

Checks that the source is valid and then runs the source

Properties

contextset value “ExecutionContext”

Uses ExecutionContext class information to provide required metadata on running source. Any stage-specific configuration resolved by the Pipeline is available through context.stage_config while this stage is running.

executorset value “StageExecutor”

Uses StageExecutor class to extract the .execute method to actually run the source.

rtype:

execute method of the StageExecutor class stored in the StageResult class.

source: Path | Callable[[...], Any] | None = None
property source_label: str | None

Sets a property for the Stage class with a human-readable name for the source.

Return type:

source_label attribute to the Stage class if source is a callable or Path.

property source_path: Path | None

Sets a property for the Stage class if the source is a path.

Return type:

source_path attribute to the Stage class if the source is a path.

validate() → None

Error checking on source attribute.

Raises:

StageConfigurationError – If source attribute does not define a source or does not exist.

with_dependencies(*dependencies: str) → Stage

Method that normalises and adds dependencies to the Stage class attributes.

Parameters:

*dependencies (str) – Information on which scripts need to run before other scripts for this Stage.

Returns:

Stage class instance with normalised dependencies attribute.

Return type:

Stage

class onsrap.StageConfig(name: str, _variables: dict[str, ~typing.Any] = <factory>, metadata: dict[str, ~typing.Any] = <factory>)

Bases: object

Holds configuration that should be exposed to an individual stage at runtime.

Parameters:

  • name (str) – The name of the stage that this configuration applies to.

  • _variables (dict[str, Any]) – Arbitrary stage-scoped variables.

  • metadata (dict[str, Any]) – Additional supporting metadata for the stage configuration.

classmethod from_mapping(name: str, data: Mapping[str, Any] | None = None) → StageConfig

Build a StageConfig from a mapping loaded from code or configuration files.

The datasets and metadata keys are extracted into their dedicated attributes. All remaining keys are treated as stage variables that should be exposed to the stage at runtime.

Parameters:

  • name (str) – Stage name that this configuration applies to.

  • data (Mapping[str, Any] or None) – Raw configuration payload for that stage.

  • parameter (# Removed global_vars)

Returns:

A normalized stage configuration object.

Return type:

StageConfig

get(variable: str, default: Any | None = None) → Any

Return a configured variable if present, otherwise return default.

get_variables(variable: Iterable[str] | str | None = None) → Any

Return all configured variables, one configured variable, or a selected subset.

metadata: dict[str, Any]
name: str
require(variable: str) → Any

Return a configured variable and raise if the stage does not define it.

to_dict() → dict[str, Any]

Serialize the stage configuration back to a mapping suitable for manifests.

property variables: dict[str, Any]

Return a copy of the stage variables without datasets or metadata.

exception onsrap.StageConfigurationError

Bases: PipelineValidationError

Raised when a stage definition is malformed. Child class with PipelineValidationError as the parent class.

exception onsrap.StageExecutionError(message: str, stage_name: str | None = None, source: str | None = None, original_exception: Exception | None = None, result: StageResult | None = None)

Bases: OnsrapError

Raised when a stage fails during execution. Child class with OnsrapError as the parent class.

class onsrap.StageExecutor(*args, **kwargs)

Bases: Protocol

Child class of Protocol Implementation required

execute(stage: Stage, context: ExecutionContext) → StageResult

Method to run Stage however implementation required

class onsrap.StageGraph(stages: list[~onsrap.stage.Stage] = <factory>)

Bases: object

Represents an order to run stages.

Holds an order that stages need to run in based on dependencies and logic.

Parameters:

stages (list of Stage class items)

classmethod from_stages(stages: Iterable[Stage]) → StageGraph

This is the primary constructor for StageGraph, which performs validation and normalization of the stage list.

Parameters:

stages (Iterable of Stage class instances)

Return type:

The stages parameter as a list.

stages: list[Stage]
topological_order() → list[Stage]

Return the stages in an order that respects their dependencies.

In graph terms, this is a topological sort: if stage B depends on stage A, then A will always appear before B in the returned list. The word “topological” here does not refer to geographic maps or terrain; it means we are arranging nodes in a dependency-safe order.

The implementation works by repeatedly selecting stages that currently have no unmet dependencies. Those stages are “ready” to run because nothing else needs to happen first. After a ready stage is placed in the output order, the algorithm removes it from the dependency lists of the stages that depend on it. That may free up more stages, which are then added to the ready list.

Returns:

  • A list of stages ordered in the way that they need to be run through the

  • pipeline.

Raises:

DependencyCycleError – If the algorithm cannot place every stage, the graph contains either a cycle or a dependency that could not be resolved.

validate() → None

Validate the stage graph for issues such as duplicate stage names, missing dependencies, and cycles.

Raises:
exception onsrap.StageLoadError(message: str, stage_name: str | None = None, source: str | None = None, original_exception: Exception | None = None, result: StageResult | None = None)

Bases: StageExecutionError

Raised when a file-backed stage cannot be loaded. Child class with StageExecutionError as the parent class.

class onsrap.StageResult(name: str, status: ~onsrap.models.StageStatus, started_at: ~datetime.datetime, finished_at: ~datetime.datetime, outputs: ~typing.Any | None = None, stdout: str = '', stderr: str = '', return_code: int | None = None, metadata: dict[str, ~typing.Any] = <factory>, error: str | None = None, source: str | None = None)

Bases: object

Holds information about how the stage ran.

Parameters:

  • name (str) – The name of the Stage run.

  • status (StageStatus) – The status of the run at completion.

  • started_at (datetime) – The date and time that the Stage started.

  • finished_at (datetime) – The date and time that the Stage finished.

  • outputs (Any, default = None) – Captures outputs of the stage being run.

  • stdout (str, default = "") – Captures outputs of the stage being run.

  • stderr (str, default = "") – Captures any errors produced during the run.

  • return_code (int, optional, default = None) – Indicates whether the stage has run successfully or if there was an error.

  • metadata (dict[str, Any]) – Holds information about the Stage such as file directories.

  • error (str, optional, default = None) – Any errors produced during the run.

  • source (str, optional, default = None) – The name/location of the code for that Stage run.

property duration_seconds: float

Creates a new attribute in the StageResult class called duration_seconds that holds the exact duration of the stage in seconds.

error: str | None = None
finished_at: datetime
metadata: dict[str, Any]
name: str
outputs: Any = None
return_code: int | None = None
source: str | None = None
started_at: datetime
status: StageStatus
stderr: str = ''
stdout: str = ''
property succeeded: bool

Creates a new attribute in the StageResult class called succeeded that contains a boolean value indicating if the run was a success or not. Updates the status attribute to record that the Stage ran successfully.

class onsrap.StageStatus(value)

Bases: str, Enum

Class to hold information on how the Stage has run.

FAILED = 'failed'
PENDING = 'pending'
RUNNING = 'running'
SKIPPED = 'skipped'
SUCCEEDED = 'succeeded'

Subpackages

Submodules