Error Handling
Error handling in UniteLabs is built into the workflow hierarchy. Different error types are handled at the level best suited to recover from them — automatic retries at the step level, and try/except with operator-assisted recovery at the phase level.
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.Error types
| Type | Examples | Where handled |
|---|---|---|
| Technical | Device timeout, lost connection, command not acknowledged | Step — automatic retry |
| Operational | Phase time limit exceeded, device unavailable at scheduling time | Phase — try/except + pause or abort |
| Scientific | Sample outside expected range, upstream result invalid | Phase — raise an exception, pause for operator intervention |
Step level: automatic retry
Steps are the first line of defense. The workflow engine retries technical errors automatically before surfacing them:
from unitelabs.sdk.automate import step
@step(retries=3, retry_delay_seconds=2)
async def aspirate(liquid_handler: LiquidHandlerDevice, volume: float):
await liquid_handler.pipettes.aspirate(volume)
If the step fails after all retries, the error is propagated to the enclosing phase.
The default retry count is configurable per step. Only retry idempotent operations — retrying a partial dispense will double-dispense already-filled wells.
Phase level: try/except and operator recovery
At the phase level, use try/except to catch errors and decide how to respond — retry, pause for operator intervention, or abort cleanly:
from unitelabs.sdk import get_logger, pause_flow_run
from unitelabs.sdk.automate import phase
class PlateNotDetectedError(Exception):
pass
@phase()
async def detection_phase(lh, carrier_id: str) -> dict:
logger = get_logger()
try:
result = await detect_plate(lh=lh, carrier_id=carrier_id)
except PlateNotDetectedError:
logger.warning(
f"No plate detected at {carrier_id}. "
"Seat the plate fully, then click Resume."
)
await pause_flow_run()
logger.info("Operator resumed — retrying detection.")
result = await detect_plate(lh=lh, carrier_id=carrier_id)
return {"lh": lh, "target_slot": result["slot"]}
Use try/finally to guarantee cleanup regardless of outcome:
@phase()
async def dispensing_phase(lh) -> dict:
try:
lh = await distribute(lh=lh, volume=100.0)
finally:
lh = await return_tips(lh=lh) # always runs — success, error, or cancellation
return {"lh": lh}
See Error recovery for more patterns.
Workflow level: propagation and setting a checkpoint
If a phase fails and no try/except handles the error, the failure propagates to the workflow. The run is marked as FAILED and the workflow engine records the last completed phase as a checkpoint.
Because phases always end in stable states, a failed run can be resumed from the last successful phase — either automatically or by an operator.
Summary
Step fails
└── Technical error → retry (up to N times)
└── Still failing → propagate to phase
Phase receives error
└── try/except → pause for operator, retry, or return error
└── No handler → propagate to workflow
Workflow receives error
└── Run marked FAILED at last stable checkpoint
└── Operator can resume from checkpoint
Related concepts
- Step: where retries are configured
- Human in the Loop: using
pause_flow_run()for operator-assisted recovery - Runs: run states and checkpoint-based resumption