Skip to content

ConduitStep

Represents a single step that is in the job.

Every step is connected to a specific function. When a step starts running, it executes the function internally. Steps also hold variables to track the execution progress of the function such as return values, statuses.

Steps can be created with create_step() in Conduit objects, but you can create a this object manually by providing a existing Conduit object.

Note

Creating a step with same ID is not allowed. However, you will not get any exception as steps are designed to be defined by users. When you run the job, it will result in DUPLICATE_STEP_IDS status code.

Attributes:

Name Type Description
job pyconduit.conduit.Conduit

An Conduit object that this step has added / attached to.

block Union[ConduitBlock, ConduitBlock._Partial]

An ConduitBlock object to specify which block will be executed.

parameters Dict[str, Any]

A dictionary of parameters that will be passed to this block as keyword parameters.

id str

An identifier for this step. It is unique among all other steps in the job. If no any ID provided when creating the step, then the position of the step will be set as step ID.

forced bool

Specifies if this step must be executed even if one (or more) of the previous steps fails. Defaults to False.

if_condition Union[str, List[str], None]

A list or string of if conditions that contains context values. These conditions will be checked before block executing starts, so if one (or more) of conditions fails, then the step will not be executed.

status ConduitStatus

It is used to store the status of step.

return_value Any

The value that returned after executing the step. If step is not executed yet, this will be None.

is_id_duplicate: bool property readonly

Check if the new step ID has used before.

Returns:

Type Description
bool

True if step ID has used more than once, False otherwise.

position: int property readonly

Returns the position of this step in job. Position index starts with 1.

Returns:

Type Description
int

An integer which represents the position index.

__init__(self, job, block, parameters={}, if_condition=None, id=None, forced=False) special

Parameters:

Name Type Description Default
job Conduit

An Conduit object that this step will added / attached to.

required
block Union[pyconduit.block.ConduitBlock, pyconduit.block.ConduitPartialBlock]

An ConduitBlock object to specify which block will be executed.

required
parameters dict

A dictionary of parameters that will be passed to this block as keyword parameters.

{}
id Optional[str]

An identifier for this step. It needs to be unique among all other steps in the job. If it is None, then the position of the step will be used as step ID.

None
forced bool

Specifies if this step must be executed even if one (or more) of the previous steps fails. Defaults to False.

False
if_condition Union[str, List[str]]

A list or string of if conditions that contains context values. These conditions will be checked before block executing starts, so if one (or more) of conditions fails, then the step will not be executed.

None

check_if_condition(self)

Checks if all if conditions passes that defined in if_condition attribute.

Returns:

Type Description
bool

True if all conditions has passed, False otherwise.

function(self)

Returns a custom function with keyword and positional parameters filled.

Returns:

Type Description
Union[Callable, Coroutine]

The custom function.

refresh_status(self)

Checks for errors and updates the step's status.

resolve_references(self)

Returns the step parameters and converts the variable references to their values.

Returns:

Type Description
Dict[str, Any]

A dictionary that contains step parameters with resolved context values.