Workflow Syntax¶
All the variables inside a workflow are of type Reference. Hence, all the arguments of the initial workflow function and the return
values of any workflow construct (task, etc.) are of type Reference.
A workflow reference (Reference) is a placeholder for objects used while a workflow is being constructed. If tasks are the nodes in the workflow graph, then references define the edges.
Reference is a proxy for the underlying object it represents within the workflow. This means that operations done
on the Reference at definition time are applied to the object only when the workflow is executed.
Variables as references¶
When a function decorated with the @workflow decorated is instantiated, the function is executed to build the workflow and the connections between the operations inside it.
Consider a workflow that takes in an argument qubit, and we print it within the workflow definition:
from laboneq import workflow
class Qubit:
def __init__(self, name: str):
self.name = name
@workflow.workflow
def a_workflow(qubit):
print(qubit)
print(qubit.name)
Now when we instantiate the workflow, the function a_workflow is executed to create the actual workflow and connections between the workflow's arguments, tasks, and operations.
As we can see below, the workflow is not yet run, yet the print() was executed at the collection phase and qubit is of type Reference, which points to the workflow argument qubit. Another Reference was created when the .name attribute was accessed.
wf = a_workflow(Qubit(name="q1"))
To keep the workflow definition simple, most regular Python operations should be done inside tasks, i.e. in functions decorated with @workflow.task. Let's see how this works.
Below, we refactor the workflow to have a task that prints the object information.
@workflow.task
def task_display_input(inp):
print("Task input:", inp)
@workflow.workflow
def a_workflow(qubit):
task_display_input(qubit)
task_display_input(qubit.name)
When instantiating the workflow, we do not see any print() messages, as the workflow and the task is not yet executed. However the workflow function was executed and we can inspect
the simple graph it made.
wf = a_workflow(Qubit(name="q1"))
wf.graph.tree
By running the workflow, the references inside it are automatically resolved when the task is called, and we should see the print() output for qubit and qubit.name:
_ = wf.run()
In the rest of the code snippets below, we will use the task_display_input(...) task defined here to print the arguments passed to it.
Constant variables¶
Variables can be defined and overwritten inside a workflow.
In the following example we define a workflow constant variable result = 5 and overwrite it if the conditional workflow.if_() returns True.
@workflow.task
def addition(x, y):
return x + y
@workflow.workflow
def a_workflow(obj):
result = 5
with workflow.if_(obj):
result = addition(1, 2)
workflow.return_(result)
Since the conditional branch is True and executed, the constant variable result is overwritten by the result of addition.
result = a_workflow(True).run()
result.output
Constant variable result is not overwritten if the conditional branching is False.
result = a_workflow(False).run()
result.output
Unresolved variables and references¶
Potentially undefined variables and unresolved references are not checked at the definition time and will fail at runtime for an unresolved value. This can happen especially in conditional branching.
Let's modify the previous example to show this. Like before, we have a conditional block that is executed only if the input is True and that sets the workflow output to the result of task addition. However, here we do not define the value of result before the conditional block, so if the condition is False, then result is not defined.
@workflow.task
def addition(x, y):
return x + y
@workflow.workflow
def a_workflow(obj):
with workflow.if_(obj):
result = addition(1, 2)
workflow.return_(result)
result = a_workflow(True).run()
result.output
As expected, when the condition is True, the workflow executes fine, because the result variable within the workflow was defined by entering the workflow.if_() block.
However, when the input is set to False, the workflow fails for unresolved Reference as the result variable never exists and is needed for workflow.result_(). workflow.result_() knows that it expects the value from addition, but it never gets it at runtime.
a_workflow(False).run()
Supported operations¶
Not all the normal Python operations are supported by the Reference object, which is mostly by design in order to keep the workflow definition simple. This is because, when an operation is done on the Reference, it produces a new Reference which then applies the operation specified during the definition phase at runtime. This means that if an invalid operation, for example, an invalid attribute is accessed, it will fail at runtime.
The supported operations on Reference is currently limited by design as workflow tasks should handle most of the operations within an workflow.
The following operations are supported:
Getting an attribute¶
@workflow.workflow
def a_workflow(obj):
task_display_input(obj.name)
_ = a_workflow(Qubit(name="q1")).run()
Getting an item¶
@workflow.workflow
def a_workflow(obj):
task_display_input(obj["qubit"])
task_display_input(obj["qubit"]["name"])
qubit_dict = {"qubit": {"name": "q1"}}
_ = a_workflow(qubit_dict).run()
Testing for equality¶
@workflow.workflow
def a_workflow(obj):
task_display_input(obj == 1)
_ = a_workflow(obj=1).run()
_ = a_workflow(obj=2).run()
Limitations¶
The Reference allows some operations to be performed on it by utilizing Python's magic methods, e.g __eq__() and so on. However, not all Python operations can be mocked in this way.
Identity operations¶
Python identity operation is should not be used at the workflow level. This is a Python limitation since it is not viable to overwrite this specific Python behaviour in objects.
In the example below, we demonstrate what happens when a workflow function uses the isstatement by applying is True on the workflow input.
@workflow.workflow
def input_is_true(obj):
task_display_input(obj is True)
_ = input_is_true(obj=True).run()
_ = input_is_true(obj=False).run()
The argument to task_display_input resolves to False no matter the value of obj. Therefore, the boolean object itself should be passed in, as shown below.
@workflow.workflow
def input_is_true(obj):
task_display_input(obj)
_ = input_is_true(obj=True).run()
_ = input_is_true(obj=False).run()
References inside containers¶
Variables and references inside a workflow should not be put into containers (list, dict, etc.), as the workflow engine needs to "see" the reference to be able to resolve it.
In the following example, we'll pass the workflow input parameter obj as a list into a task argument.
@workflow.workflow
def input_is_true(obj):
task_display_input([obj])
_ = input_is_true(obj=1).run()
As we can see from the output, the task got an argument of type list, which contains a Reference, not the expected [1].
However by passing obj=[1] to the workflow input, we obtain the expected behaviour and our task gets the correct value:
@workflow.workflow
def input_is_true(obj):
task_display_input(obj)
_ = input_is_true(obj=[1]).run()
Getting around the limitations¶
As there are operations that are not supported on the workflow level, we can implement them inside tasks.
In the following example we create a helper task, which converts its input into a boolean. We also configure the task with save=False, which signals to the LogBook to not save it's input and output on disk.
@workflow.task(save=False)
def is_true(x):
return bool(x)
@workflow.workflow
def input_is_true(obj):
task_display_input(is_true(obj))
_ = input_is_true(obj=True).run()
_ = input_is_true(obj=False).run()
_ = input_is_true(obj=[]).run()
_ = input_is_true(obj=[1]).run()
As a final example, let's define a helper task that appends entries to a list.
@workflow.task
def append_things(things_list, thing):
things_list += [thing]
return things_list
@workflow.task
def append_things_dynamic(things_list, thing):
things_list += [thing]
@workflow.workflow
def workflow_to_append_things():
lst = []
with workflow.for_(list(range(6))) as i:
lst = append_things(lst, i)
append_things_dynamic(lst, i)
workflow.return_(lst)
result = workflow_to_append_things().run()
result.output
Notice that both returning the extended list (append_things) and modifying it in place inside the task (append_things_dynamic) leads to the same correct behaviour.