onsrap.execution module

class onsrap.execution.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.execution.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.execution.StageExecutor(*args, **kwargs)

Bases: Protocol

Child class of Protocol Implementation required

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

Method to run Stage however implementation required