Module: workflow

This module contains the implementation of the base components which are Workflow and Step classes.

Step

A class to represent a step instance that performs a specific task in a workflow.

Parameters:

Name	Type	Description	Default
name	str	The name of the step (it's ID).	''

Attributes:

Name	Type	Description
logger	Logger	The logger of the step.
workflow_context	WorkflowContext	The context of the workflow.
path	StepPath	The path to the step.
name	str	The name of the step (it's ID).

default_logger_name: str property

A property to get the default name of the logger that is composed of the default logger prefix and the name of the step.

information: StepInformation property

A property to get the information about the step.

configure_logger()

A method to configure the logger of the step.

fail()

A method to set the status of the step to 'failed'.

perform(*args, **kwargs)

A method to perform the task of the step.

Parameters:

Name	Type	Description	Default
args		The positional arguments of the step.	()
kwargs		The keyword arguments of the step.	{}

Returns:

Type	Description
Any	The return value of the step.

Raises:

Type	Description
NotImplementedError	The method 'perform' must be implemented in the child class.

success()

A method to set the status of the step to 'success'.

StepInformation dataclass

A class to represent the information of a step in a workflow.

Parameters:

Name	Type	Description	Default
path	StepPath	The path to the step.	required
status	StepStatus	The status of the step.	required
parameters	Optional[Dict]	The parameters of the step.	None
stdout	Optional[str]	The standard output of the step.	None
stderr	Optional[str]	The standard error of the step.	None
error	Optional[Exception]	The error (exception) of the step.	None
return_value	Optional[Any]	The return value of the step.	None
previous_step	Optional[StepInformation]	The previous step in the workflow.	None
next_step	Optional[StepInformation]	The next step in the workflow.	None
parent	Optional[StepInformation]	The parent step in the workflow.	None
children	Optional[List[StepInformation]]	The children steps in the workflow.	None

to_dict()

A method to convert the step information to a dictionary.

Returns:

Type	Description
List[Dict]	The step information as a dictionary.

StepPath

A class to represent the path to a step in a workflow.

name: str property writable

A property to get the name of the step.

type: StepType property

A property to get the type of the step.

StepStatus

Bases: Enum

A class to represent the status of a step in a workflow.

Steps

A class to manage the steps in the workflow.

register(name)

A method to register a step in the workflow.

Parameters:

Name	Type	Description	Default
name	str	The name of the step.	required

Returns:

Type	Description
Callable[[Type[Step]], None]	The class wrapper.

wrap_step() staticmethod

A method to wrap the step.

Parameters:

Name	Type	Description	Default
self	Step	The step to wrap.	required

Returns:

Type	Description
	The wrapped step.

StepsInformation dataclass

A class to represent the information of all steps in a workflow.

Parameters:

Name	Type	Description	Default
steps	Dict[StepPath, StepInformation]	The dictionary of steps in the workflow.	dict()

first_step: Optional[StepInformation] property

A property to get the first step in the workflow.

Returns:

Type	Description
Optional[StepInformation]	The first step in the workflow.

get_step_information(step_path)

A method to get the status of a step in the workflow.

Parameters:

Name	Type	Description	Default
step_path	StepPath	The path to the step.	required

Returns:

Type	Description
StepInformation	The status of the step.

to_dict()

A method to convert the steps information to a dictionary.

Returns:

Type	Description
Dict	The steps information as a dictionary.

WorkflowContext

A class to represent the context of a workflow. Context is a dictionary that stores the state of the workflow that is shared between steps.

Parameters:

Name	Type	Description	Default
parameters	Optional[Dict]	The parameters of the workflow.	None
steps_information	Optional[StepsInformation]	The status of the steps in the workflow.	None

global_lock: Lock property

A property to get the global lock of the workflow.

steps_information: StepsInformation property

A property to get the status of all steps in the workflow.

get(key, default=None)

A method to get a value from the context.

Parameters:

Name	Type	Description	Default
key	str	The key of the value to get from the context.	required
default	Any	The default value to return if the key is not found in the context.	None

Returns:

Type	Description
Any	Value from the context, if the key is found, otherwise the default value.

get_step_information(step)

A method to get the status of a step in the workflow.

Parameters:

Name	Type	Description	Default
step	StepPath	The path to the step.	required

Returns:

Type	Description
StepInformation	The status of the step.

set(key, value)

A method to set a value to the context.

Parameters:

Name	Type	Description	Default
key	str	The key of the value to set in the context.	required
value	Any	The value to set in the context.	required