Step

Pipeline step data model.

Defines Step, the unit of execution inside a Kale Pipeline, together with StepConfig and the PipelineParam / Artifact named tuples used to describe step inputs and outputs.

class kale.step.PipelineParam(param_type: str, param_value: Any)

Bases: NamedTuple

A pipeline parameter.

param_type: str

Alias for field number 0

param_value: Any

Alias for field number 1

class kale.step.Artifact(name: str, type: str, is_input: bool = False)

Bases: NamedTuple

A Step artifact.

name: str

Alias for field number 0

type: str

Alias for field number 1

is_input: bool

Alias for field number 2

class kale.step.StepConfig(*args, **kwargs)

Bases: Config

Config class used for the Step object.

name = <kale.config.config.Field object>

labels = <kale.config.config.Field object>

annotations = <kale.config.config.Field object>

limits = <kale.config.config.Field object>

base_image = <kale.config.config.Field object>

enable_caching = <kale.config.config.Field object>

generate_html_report = <kale.config.config.Field object>

retry_count = <kale.config.config.Field object>

retry_interval = <kale.config.config.Field object>

retry_factor = <kale.config.config.Field object>

retry_max_interval = <kale.config.config.Field object>

timeout = <kale.config.config.Field object>

class kale.step.Step(source: list[str] | Callable, ins: list[Any] = None, outs: list[Any] = None, **kwargs)

Bases: object

Class used to store information about a Step of the pipeline.

artifacts: list[Artifact]

parameters: dict[str, PipelineParam]

__call__(*args, **kwargs)

Handler for when the @step decorated function is called.

add_artifact(artifact_name, artifact_type, is_input)

Helper method to add an artifact to the step.

Artifact_type will be either ‘Dataset’, ‘Model’, ‘HTML’, ‘Metrics’, ‘ClassificationMetrics’ or ‘Artifact’. This will simplify tracking what should be an Input[Artifact] or Output[Artifact].

Parameters:

• artifact_name (str) – Name of the artifact.

• artifact_type (str) – Type of the artifact.

• is_input (bool) – Whether the artifact is an input or output.

run(pipeline_parameters_values: dict[str, PipelineParam])

Run the step locally.

property name

Get the name of the step.

merge_code(source_code: str)

Add a new code block to the step.

Parameters:

source_code (str) – Python source code to be appended to step

property pps_names

Get the names of the step’s parameters sorted.

property pps_types

Get the types of the step’s parameters, sorted by name.

property pps_values

Get the values of the step’s parameters, sorted by name.

property rendered_source

Source to be rendered in the template.

property kfp_inputs: list[PipelineParam | Artifact]

Get the inputs of the step for KFP.

This combines PipelineParams and Artifacts marked as inputs. Add PipelineParams first (as they’re usually positional/keyword args)

property kfp_outputs: list[Artifact]

Get Artifacts that are outputs.

class kale.step.SubPipeline(pipeline, notebook_path: str, source: list[str] = None, ins: list[Any] = None, outs: list[Any] = None, display_name: str = None, imports_and_functions: str = '', **kwargs)

Bases: Step

A referenced notebook, compiled as a nested pipeline.

A node of the same graph as Step, carrying the same name, ins, outs and config, so dependency detection and code generation treat a notebook reference and a plain step uniformly.

source is the referenced notebook’s step code: the boundary scanner reads it exactly as it reads a step’s, which is what lets a variable cross a notebook boundary the same way it crosses a step boundary. pipeline is the processed child Pipeline that becomes the nested sub-DAG.

produced_by: dict[str, str]

property display_name

Name shown for the nested pipeline, defaulting to a KFP-safe name.

merge_code(source_code: str)

A notebook reference never absorbs the code of another cell.

run(pipeline_parameters_values: dict[str, PipelineParam])

Run the referenced notebook’s steps locally, in order.

kale.step.execution_handler(step: Step, *args, **kwargs)