onsrap.models module

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

Bases: object

contents: dict[str, Any]
description: str
name: str
class onsrap.models.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.models.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.models.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.models.PipelineStatus(value)

Bases: str, Enum

Class to hold information on how the Pipeline has run.

FAILED = 'failed'
PENDING = 'pending'
RUNNING = 'running'
SUCCEEDED = 'succeeded'
class onsrap.models.RAPDataset

Bases: object

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

class onsrap.models.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.models.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'
onsrap.models.now() → datetime

Function to extract the current time in a datetime format.

onsrap.models.utcnow() → datetime

Function to extract the current time in UTC in a datetime format.