laboneq.workflow.core

Core workflow objects.

Parameters = ParamSpec('Parameters') module-attribute

Workflow(root, input=None)

Bases: Generic[Parameters]

Workflow for task execution.

Parameters:

Name	Type	Description	Default
root	WorkflowBlock	A root workflow block.	required
input	dict | None	Input parameters of the workflow.	None

graph: WorkflowGraph property

Workflow graph object.

input: dict property

Input parameters of the workflow.

name: str property

Workflow name.

from_callable(func, name=None, input=None) classmethod

Create a workflow from a callable.

Parameters:

Name	Type	Description	Default
func	Callable	A callable defining the workflow	required
name	str | None	Name of the workflow	None
input	dict | None	Input parameters of the workflow	None

resume(until=None)

Resume workflow execution.

Resumes the workflow execution from the previous state.

Parameters:

Name	Type	Description	Default
until	str | None	Run until a first task with the given name. None will fully execute the workflow. Until cannot be used for tasks and sub-workflows inside loops.	None

Returns:

Type	Description
WorkflowResult	Result of the workflow execution.
WorkflowResult	if until is used, returns the results up to the selected task or sub-workflow.

Raises:

Type	Description
WorkflowError	An error occurred during workflow execution or workflow is not in progress.
ValueError	'until' value is invalid.

run(until=None)

Run the workflow.

Resets the state of an workflow before execution.

Parameters:

Name	Type	Description	Default
until	str | None	Run until the first task or sub-workflow with the given name. None will fully execute the workflow. Until cannot be used for tasks and sub-workflows inside loops. If until is used, the workflow execution can be resumed with .resume().	None

Returns:

Type	Description
WorkflowResult	Result of the workflow execution.
WorkflowResult	if until is used, returns the results up to the selected task.

Raises:

Type	Description
WorkflowError	An error occurred during workflow execution.
ValueError	'until' value is invalid.

WorkflowBuilder(func, name=None)

Bases: Generic[Parameters]

A workflow builder.

Builds a workflow out of the given Python function.

Parameters:

Name	Type	Description	Default
func	Callable[Parameters]	A python function, which acts as the core of the workflow.	required
name	str | None	Name of the workflow. Defaults to wrapped function name.	None

src: str property

Source code of the workflow.

options()

Create default options for the workflow.

The option attribute tasks is populated with all the sub-task and sub-workflow options within this workflow.

recover()

Recover the result of the last run to raise an exception.

Returns the result of the last failed run of a workflow created from this workflow builder. In no run has failed, an exception is raised.

After a result is recovered, the result is cleared and further calls to .recover will raise an exception.

Returns:

Type	Description
WorkflowResult	Latest workflow that raised an exception.

Raises:

Type	Description
WorkflowError	Raised if no previous run failed.

workflow(func=None, name=None)

A decorator to mark a function as workflow.

The arguments of the function will be the input values for the wrapped function.

If workflow decorated function is called within another workflow definition, it adds a sub-graph to the workflow being built.

Parameters:

Name	Type	Description	Default
func	Callable[Parameters] | None	A function that defines the workflow structure. The arguments of func can be freely defined, except for an optional argument options, which must have a type hint that indicates it is of type WorkflowOptions or its' subclass, otherwise an error is raised.	None
name	str | None	Name of the workflow. Defaults to wrapped function name.	None

Returns:

Type	Description
WorkflowBuilder[Parameters] | Callable[[Callable[Parameters]], WorkflowBuilder[Parameters]]	A wrapper which returns a Workflow instance if called outside of another workflow definition.
WorkflowBuilder[Parameters] | Callable[[Callable[Parameters]], WorkflowBuilder[Parameters]]	A wrapper which returns a WorkflowResult if called within another workflow definition.

Example

from laboneq import workflow


@workflow.workflow
def my_workflow(x: int): ...


wf = my_workflow(x=123)
results = wf.run()