Conditions & Error Handling
on_success, on_failure, always, callable conditions, fail_fast vs continue, skip propagation.
on_success, on_failure, always, callable conditions, fail_fast vs continue, skip propagation.
Control which steps execute based on predecessor outcomes, and configure how the workflow responds to failures.
wf.step("deploy", deploy, after="test") # default: on_success
wf.step("rollback", rollback, after="deploy", condition="on_failure")
wf.step("notify", send_slack, after="deploy", condition="always")| Condition | Runs when |
|---|---|
None / "on_success" | All predecessors completed successfully |
"on_failure" | Any predecessor failed |
"always" | Predecessors are terminal (regardless of outcome) |
callable | condition(ctx) returns True |
Pass a function that receives a WorkflowContext:
from flexiq.workflows import WorkflowContext
def high_score(ctx: WorkflowContext) -> bool:
return ctx.results["validate"]["score"] > 0.95
wf.step("deploy", deploy, after="validate", condition=high_score)WorkflowContext fields:
| Field | Type | Description |
|---|---|---|
run_id | str | Workflow run ID |
results | dict[str, Any] | Deserialized return values of completed nodes |
statuses | dict[str, str] | Status strings for all terminal nodes |
failure_count | int | Number of failed nodes |
success_count | int | Number of completed nodes |
Set the workflow-level error strategy. Don't confuse this with the
step-level condition from the previous section: Workflow(on_failure=...)
picks the engine's global strategy ("fail_fast" or "continue"), while
condition="on_failure" on a step is a per-step predicate meaning "run this
step if a predecessor failed" — same word, different axis.
wf = Workflow(name="strict", on_failure="fail_fast")One failure skips all pending steps. The workflow transitions to FAILED.
wf = Workflow(name="resilient", on_failure="continue")Failed steps skip their on_success dependents, but independent
branches keep running.
When a node is skipped, its successors are evaluated recursively:
on_success successors → SKIPPED (predecessor didn't succeed)on_failure successors → evaluated (predecessor is terminal)always successors → run regardless of how the predecessor endedwf = Workflow(name="cleanup_pipeline")
wf.step("a", risky_task)
wf.step("b", next_step, after="a") # SKIPPED if a fails
wf.step("cleanup", cleanup, after="b", condition="always") # runs even if b is skippedConditions work with fan-out nodes. If a fan-out child fails:
wf.step("fetch", fetch_data)
wf.step("process", process, after="fetch", fan_out="each")
wf.step("aggregate", aggregate, after="process", fan_in="all")
wf.step("on_error", alert, after="process", condition="on_failure")If any process[i] child fails, the fan-out parent is marked FAILED,
aggregate is skipped, and on_error runs.