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>

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.

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