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 property

Workflow graph object.

input property

Input parameters of the workflow.

name 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 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)

workflow(func: Callable[Parameters], name: str) -> WorkflowBuilder[Parameters]

workflow(func: Callable[Parameters]) -> WorkflowBuilder[Parameters]

workflow(func: None = ..., name: str | None = ...) -> Callable[[Callable[Parameters]], WorkflowBuilder[Parameters]]

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()