laboneq.workflow.blocks

Workflow block definitions.

Block(parameters=None)

Bases: ABC

A base class for workflow blocks.

A block can be an individual task or a collection of other blocks.

Classes inheriting from Block must define the following methods:

- `execute()`: A method that executes the block and it's children defined
    in `Block.body`.

Parameters:

Name	Type	Description	Default
parameters	dict | None	Expected input parameters of the block.	None

body: list[Block] property

Body of the block.

A list of other blocks that are defined within this block.

hidden: bool property

Whether or not the block is a hidden block.

Hidden blocks are generally used only for execution and they are not deemed relevant in results.

name: str property

Name of the block.

parameters: dict property

Input parameters of the block.

collect()

Collect blocks defined within the context.

Every block defined within the context are added to the body of the block.

execute(executor) abstractmethod

Execute the block.

extend(blocks)

Extend the body of the block.

iter_child_blocks()

Iterate over the children of this block.

BlockBuilderContext

Bases: LocalContext[Block]

Workflow block builder context.

BlockVisitor()

A block tree visitor.

This class is meant to be subclassed.

Subclasses can implement custom visitor method for each block type. The format for naming the methods are visit_<block type lowercase>

For example, to write a visitor method for `WorkflowBlock`:

    `def visit_workflowblock(self, block):`

The visitor method should visit or call .generic_visit() to visit the children.

generic_visit(block)

Visit block children.

Parameters:

Name	Type	Description	Default
block	Block	Root block.	required

visit(block)

Visit a block and its children recursively.

Parameters:

Name	Type	Description	Default
block	Block	Root block.	required

BreakLoopBlock(parameters=None)

Bases: Block

Break loop block.

A block that breaks out of the currently running innermost workflow loop.

execute(executor)

Execute the block.

ForExpression(values, loop_indexer=None)

Bases: Block

For expression.

A block that iterates workflow blocks over the given values.

The expression will always fully execute regardless if the workflow is partially executed or not.

Parameters:

Name	Type	Description	Default
values	Iterable | Reference	An iterable. Iterable can contain workflow objects.	required
loop_indexer	Callable[[object], object] | None	A callable to produce dynamic loop indexes for each item. The callable must accept a single argument, the item currently being iterated. The index should be a hashable and unique across the items in the loop. Usually they are simple Python types such as str, int or tuples of these. When loop_indexer is None, each task executed within the loop is given a loop index from 0..len(values) - 1.	None

ref: Reference property

Reference to the object.

execute(executor)

Execute the block.

IFExpression(condition)

Bases: Block

If expression.

A block that is executed if a given condition is true.

Parameters:

Name	Type	Description	Default
condition	object	A condition that has to be true for block to be executed.	required

execute(executor)

Execute the block.

Namespace(**kwargs)

A class to provide attribute access to an immutable namespace.

Parameters:

Name	Type	Description	Default
**kwargs	object	Attributes to set to the namespace.	{}

ReturnStatement(callback, /, **kwargs)

Bases: Block

Return statement for a workflow.

Sets the active workflow block output and interrupts the current workflow execution.

Parameters:

Name	Type	Description	Default
callback	Callable	Callback producing the value to be set for the workflow output.	required
**kwargs	object	Keyword arguments for the callback.	{}

callback = callback instance-attribute

execute(executor)

Execute the block.

from_value(value) classmethod

Construct a return statement that returns the just given value.

Parameters:

Name	Type	Description	Default
value	object	The value to be returned as the workflow output.	required

Returns:

Name	Type	Description
block	ReturnStatement	A return block.

TaskBlock(task, parameters=None)

Bases: Block

Task block.

TaskBlock is an workflow executor for a task.

Parameters:

Name	Type	Description	Default
task	task_	A task this block contains.	required
parameters	dict | None	Input parameters of the task.	None

hidden: bool property

Whether or not the task is a hidden task.

name: str property

Name of the task.

options_type: type[TaskOptions] | None property

Type of block options.

ref: Reference property

Reference to the object.

task = task instance-attribute

execute(executor)

Execute the task.

WorkflowBlock(name, options_type=WorkflowOptions, parameters=None)

Bases: Block

Workflow block.

name: str property

Name of the block.

options_type: type[WorkflowOptions] property

Type of workflow options.

ref: Reference property

Reference to the object.

create_options()

Create options for the block.

The method goes over the sub-blocks and finds which blocks has options available.

Same task option instance is shared across the same named task executions per workflow, therefore the same task calls within a single workflow block cannot have multiple option definitions.

Returns:

Type	Description
WorkflowOptions	Workflow options where tasks is populated with the default options of sub-blocks.

execute(executor)

Execute the block.

from_callable(name, func, **kwargs) classmethod

Create the block from a callable.

By default the signature of the function is used to define the default parameters of the block.

Parameters:

Name	Type	Description	Default
name	str	Name of the block.	required
func	Callable	A function defining the workflow	required
**kwargs	object	Default parameter values that overwrite function signature	{}

set_params(executor, **kwargs)

Set the initial parameters of the block.

Parameters:

Name	Type	Description	Default
executor	ExecutorState	Active executor.	required
**kwargs	object	Input parameters of the block.	{}

break_()

Break statement to break out of the currently running innermost loop.

The equivalent of Python's break statement.

Raises:

Type	Description
WorkflowError	Defined outside of loop scope.

elif_(condition)

Workflow else if statement.

The equivalent of Python's elif-statement.

Parameters:

Name	Type	Description	Default
condition	Any	A condition that has to be true for code block to be executed.	required

Example

from laboneq import workflow


@workflow.workflow
def a_workflow(x):
    with workflow.if_(x == 1):
        ...
    with workflow.elif_(x == 2):
        ...

Raises:

Type	Description
WorkflowError	Expression is defined without if_()

else_()

Workflow else statement.

The equivalent of Python's else-statement.

Example

from laboneq import workflow


@workflow.workflow
def a_workflow(x):
    with workflow.if_(x == 1):
        ...
    with workflow.elif_(x == 2):
        ...
    with workflow.else_():
        ...

Raises:

Type	Description
WorkflowError	Expression is defined without elif_() or if_()

for_(values, loop_indexer=None)

For expression to iterate over the values within a code block.

The equivalent of Python's for loop.

Parameters:

Name	Type	Description	Default
values	Iterable[T]	An iterable.	required
loop_indexer	Callable[[object], str] | None	A callable to produce dynamic loop indexes for each item. The callable must accept a single argument, the item currently being iterated. The index should be a hashable and unique across the items in the loop. Usually they are simple Python types such as str, int or tuples of these. When loop_indexer is None, each task executed within the loop is given a loop index from 0..len(values) - 1.	None

Example

with workflow.for_([1, 2, 3]) as x:
    ...

if_(condition)

Workflow if statement.

The equivalent of Python's if-statement.

Parameters:

Name	Type	Description	Default
condition	Any	A condition that has to be true for code block to be executed.	required

Example

from laboneq import workflow


@workflow.workflow
def a_workflow(x):
    with workflow.if_(x == 1):
        ...

return_(output=None, /, **kwargs)

Return statement of an workflow.

Interrupts the current workflow execution and sets the active workflow output value either to the given output value or to a Namespace object by using the given keyword arguments.

The equivalent of Python's return statement.

Parameters:

Name	Type	Description	Default
output	Any | None	Value to be set for workflow output.	None
**kwargs	object	Keyword arguments that are passed to the Namespace object.	{}

Raises:

Type	Description
TypeError	Function is called with positional and keyword arguments.