Workflow
A workflow is the top-level executable process designed to achieve a specific scientific outcome. It workflow engine phases, manages data flow between them, and owns the state of all samples and resources for the duration of the run.
You define a workflow as a Python function decorated with @workflow. Inside it, you call phases like regular functions — the workflow engine handles scheduling, parallelism, error recovery, and setting a checkpoint .
unitelabs-sdk< 0.10.0, import flow and task from Prefect instead. Use flow in place of both workflow and phase decorators, and task as step decorators.Example
from unitelabs.sdk.automate import workflow
from .phases import sample_preparation, agitation, washing_cycle, detection
@workflow(name="ELISA")
async def main_workflow():
plate = await sample_preparation()
await agitation(plate)
for _ in range(3):
await washing_cycle()
await detection()
The workflow drives the scientific narrative. Phases represent the meaningful stages — sample_preparation, agitation, detection — while the workflow defines the order and control flow between them.
Key properties
- Scientific goal: a workflow is defined by what result it produces (e.g., "ELISA"), not the hardware it uses
- Control flow: supports loops, conditionals, and branching; not limited to a linear sequence
- State ownership: is the source of truth for sample identity, lineage, and all consumed resources across phases
- Versioned: workflows are deployed and versioned; each run is linked to a specific version
Non-linear control flow
Because a workflow is just a Python function, you can use standard control flow:
@workflow(name="Repeat Until Clean")
async def repeat_workflow():
result = await initial_wash()
while not result.is_clean:
result = await additional_wash()
await final_rinse()
Parallel phases
Phases with no dependency between them can run in parallel. The workflow engine detects independence automatically — you only need to express the data dependency:
@workflow(name="Parallel Preparation")
async def parallel_workflow():
# These three phases have no shared inputs — the workflow engine runs them concurrently
reagent_a = await prepare_reagent_a()
reagent_b = await prepare_reagent_b()
await wash_plate()
# This phase depends on both reagents, so it waits for both to complete
await combine(reagent_a, reagent_b)