Control Flow

Control Flow Patterns

The workflows library provides four essential control flow patterns to structure your automation logic.

Sequential

Execute steps one after another in a defined order

Conditionals

Branch execution based on runtime conditions

Loops

Process collections of items with repeated logic

Parallel

Execute multiple branches simultaneously

Sequential

Sequential execution runs steps one after another in order. Use the .then() method to chain steps together.

from workflows_py.workflow import Step, Workflow

def step_one(ctx):
    data = ctx.get_start_data()
    return {"message": f"Hello {data['name']}", "count": 1}

def step_two(ctx):
    result = ctx.get_step_result("step-1")
    return {"message": result["message"] + "!", "count": result["count"] + 1}

def step_three(ctx):
    result = ctx.get_step_result("step-2")
    return {"final_message": result["message"], "total_count": result["count"] + 1}

workflow = Workflow()
workflow.then(
    Step("step-1", step_one)
).then(
    Step("step-2", step_two)
).then(
    Step("step-3", step_three)
)

Each step can access the results of previous steps using ctx.get_step_result("step-id").

Conditionals

Conditional branching lets your workflow make different choices based on data. Use .if_condition(), .elif_branch(), and .else_branch() to create decision logic.

from workflows_py.workflow import Step, Workflow

def check_number(ctx):
    number = ctx.get_start_data()["value"]
    return {"number": number}

def handle_positive(ctx):
    return {"result": "Number is positive"}

def handle_negative(ctx):
    return {"result": "Number is negative or zero"}

workflow = Workflow()
workflow.then(
    Step("check-number", check_number)
).if_condition(
    condition=lambda ctx: ctx.get_step_result("check-number")["number"] > 0,
    branch_fn=lambda w: w.then(
        Step("positive-path", handle_positive)
    )
).else_branch(
    branch_fn=lambda w: w.then(
        Step("negative-path", handle_negative)
    )
)

from workflows_py.workflow import Step, Workflow

def check_score(ctx):
    score = ctx.get_start_data()["score"]
    return {"score": score}

def grade_a(ctx):
    return {"grade": "A", "message": "Excellent!"}

def grade_b(ctx):
    return {"grade": "B", "message": "Good job!"}

def grade_c(ctx):
    return {"grade": "C", "message": "Needs improvement"}

workflow = Workflow()
workflow.then(
    Step("check-score", check_score)
).if_condition(
    condition=lambda ctx: ctx.get_step_result("check-score")["score"] >= 90,
    branch_fn=lambda w: w.then(Step("grade-a", grade_a))
).elif_branch(
    condition=lambda ctx: ctx.get_step_result("check-score")["score"] >= 80,
    branch_fn=lambda w: w.then(Step("grade-b", grade_b))
).else_branch(
    branch_fn=lambda w: w.then(Step("grade-c", grade_c))
)

from workflows_py.workflow import Step, Workflow

def check_number(ctx):
    number = ctx.get_start_data()["value"]
    return {"number": number}

def handle_positive(ctx):
    return {"result": "Number is positive"}

def handle_negative(ctx):
    return {"result": "Number is negative or zero"}

def final_step(ctx):
    # If positive-path was not executed, then get_step_result will return None
    result = ctx.get_step_result("positive-path") or ctx.get_step_result("negative-path")
    return {"final_result": result}

workflow = Workflow()
workflow.then(
    Step("check-number", check_number)
).if_condition(
    condition=lambda ctx: ctx.get_step_result("check-number")["number"] > 0,
    branch_fn=lambda w: w.then(
        Step("positive-path", handle_positive)
    )
).else_branch(
    branch_fn=lambda w: w.then(
        Step("negative-path", handle_negative)
    )
).then(
    Step("final-step", final_step)
)

Make sure your condition functions handle missing or invalid data gracefully.

Loops

Loops process a list of items by running the same workflow logic for each item. Use ctx.scope["item"] to access the current item and ctx.scope["i"] to get the index (0-based).

from workflows_py.workflow import Step, Workflow

def get_numbers(ctx):
    return [10, 20, 30, 40, 50]

def process_number(ctx):
    # Access the current item and its index
    current_number = ctx.scope["item"]
    current_index = ctx.scope["i"]

    return {
        "index": current_index,
        "original": current_number,
        "doubled": current_number * 2
    }

def log_result(ctx):
    current_number = ctx.scope["item"]
    result = ctx.get_step_result("process-number")

    return {
        "message": f"Item {result['index']}: {result['original']} -> {result['doubled']}"
    }

workflow = Workflow()
workflow.then(
    Step("get-numbers", get_numbers)
).loop(
    "process-numbers",
    get_items=lambda ctx: ctx.get_step_result("get-numbers"),
    branch_fn=lambda w: w.then(
        Step("process-number", process_number)
    ).then(
        Step("log-result", log_result)
    )
)

from workflows_py.workflow import Step, Workflow

def get_users(ctx):
    return [
        {"name": "Alice", "age": 25},
        {"name": "Bob", "age": 30},
        {"name": "Charlie", "age": 35}
    ]

def process_user(ctx):
    user = ctx.scope["item"]
    index = ctx.scope["i"]

    return {
        "user_id": f"user_{index}",
        "name": user["name"],
        "age": user["age"],
        "status": "active"
    }

def send_welcome_email(ctx):
    user = ctx.scope["item"]
    processed_user = ctx.get_step_result("process-user")

    return {
        "email_sent": True,
        "recipient": user["name"],
        "subject": f"Welcome {user['name']}!"
    }

workflow = Workflow()
workflow.then(
    Step("get-users", get_users)
).loop(
    "process-users",
    get_items=lambda ctx: ctx.get_step_result("get-users"),
    branch_fn=lambda w: w.then(
        Step("process-user", process_user)
    ).then(
        Step("send-welcome-email", send_welcome_email)
    )
)

Loop iterations run in parallel by default.

Batching Loop Iterations

Because loop iterations run in parallel, looping over a large list can fire many API calls at once. When the work inside the loop is rate limited (for example, an external integration or a high-volume endpoint), a common pattern is to loop over batches instead of individual items, and stagger each batch with a delay based on its index.

from workflows_py.workflow import Step, Workflow
from client_manager import SampleHealthcareClient

BATCH_SIZE = 25
BATCH_GAP_MINUTES = 5

def make_batches(ctx):
    items = ctx.get_step_result("fetch-items")
    return [items[i : i + BATCH_SIZE] for i in range(0, len(items), BATCH_SIZE)]

def process_batch(ctx):
    batch = ctx.scope["item"]
    batch_index = ctx.scope["i"]
    client = SampleHealthcareClient()

    # Each batch waits (batch_index * BATCH_GAP_MINUTES) before processing,
    # so batches are spread out instead of all running at once.
    if not ctx.resume_data and batch_index > 0:
        delay_ms = batch_index * BATCH_GAP_MINUTES * 60 * 1000
        sleep_result = client.v2.async_results.sleep(delay=delay_ms)
        client.v2.workflow_runs.resume_when_complete(
            async_result_id=sleep_result.async_result_id,
        )
        ctx.suspend_fn({"type": "staggered_batch_delay", "batch_index": batch_index})
        return None

    return [process_item(item) for item in batch]

workflow = Workflow()
workflow.then(
    Step("fetch-items", fetch_items)
).then(
    Step("make-batches", make_batches)
).loop(
    "process-batches",
    get_items=lambda ctx: ctx.get_step_result("make-batches"),
    branch_fn=lambda w: w.then(
        Step("process-batch", process_batch)
    )
)

Batch 0 processes immediately, batch 1 after one gap, batch 2 after two gaps, and so on. Suspending during the delay is durable, so this works for gaps of minutes or hours. See Delays and Scheduled Resumes for how the sleep pattern works.

Parallel

Parallel execution runs multiple workflow branches at the same time. All branches must complete before the workflow continues to the next step.

from workflows_py.workflow import Step, Workflow

def fetch_user_data(ctx):
    user_id = ctx.get_start_data()["user_id"]
    return {"name": "John", "email": "john@example.com"}

def fetch_user_orders(ctx):
    user_id = ctx.get_start_data()["user_id"]
    return {"orders": [{"id": 1, "total": 50}, {"id": 2, "total": 75}]}

def fetch_user_preferences(ctx):
    user_id = ctx.get_start_data()["user_id"]
    return {"theme": "dark", "notifications": True}

def combine_results(ctx):
    user_data = ctx.get_step_result("fetch-user-data")
    orders = ctx.get_step_result("fetch-user-orders")
    preferences = ctx.get_step_result("fetch-user-preferences")

    return {
        "user": user_data,
        "orders": orders["orders"],
        "preferences": preferences,
        "total_orders": len(orders["orders"])
    }

workflow = Workflow()
workflow.parallel(
    "gather-user-info",
    branches_fn=[
        lambda w: w.then(Step("fetch-user-data", fetch_user_data)),
        lambda w: w.then(Step("fetch-user-orders", fetch_user_orders)),
        lambda w: w.then(Step("fetch-user-preferences", fetch_user_preferences)),
    ]
).then(
    Step("combine-results", combine_results)
)

Parallel branches are perfect for independent tasks that can run simultaneously.

Retries

You can configure automatic retries for individual steps using the options parameter on Step.

from workflows_py.workflow import Step, Workflow

def unstable_operation(ctx):
    # Your step logic that may raise
    return {"ok": True}

workflow = Workflow()
workflow.then(
    Step(
        "unstable-operation",
        unstable_operation,
        options={
            "num_retries": 3,  # retry up to 3 times on failure
            "backoff": 1000,   # start with 1000ms delay; doubles each retry
        },
    )
)

num_retries: Number of retry attempts after the initial failure. Default is 0 (no retries).
backoff: Initial delay in milliseconds before the first retry. The delay doubles after each failed attempt (exponential backoff). Default is 1000ms. Must be >= 0.
Partial config: You may specify only one key:
- If you set only num_retries, the backoff defaults to 1000ms.
- If you set only backoff, retries will not happen unless num_retries > 0.

Available in workflows-py >= 0.1.21.

Error Handling

You can define custom error handling logic for individual steps using the on_failure option. This handler is called when a step fails after all retry attempts have been exhausted.

from workflows_py.workflow import Step, Workflow

def process_payment(ctx):
    # This step might fail
    raise Exception("Payment gateway unavailable")

def handle_payment_failure(ctx, error):
    # Custom error handling logic
    print(f"Payment failed: {error}")
    # You could notify admins, log to external service, etc.

workflow = Workflow()
workflow.then(
    Step(
        "process-payment",
        process_payment,
        options={
            "on_failure": handle_payment_failure,
        },
    )
)

The on_failure handler receives two arguments:

ctx: The WorkflowRunContext with access to get_step_result(), get_start_data(), etc.
error: The error information (may be None in some cases)

Combine on_failure with num_retries to first attempt automatic recovery, then execute custom logic if all retries fail.

The on_failure handler must accept exactly two arguments: (ctx, error). A handler with only one argument will raise a TypeError.

Available in workflows-py >= 0.1.28.

Next Task Routing

You can override which task Sample opens next after a step completes by using the next_task option. This is useful when a workflow fans out into multiple manual tasks and you want to send the user to a specific one, such as the most recently created task in the workflow run or the latest task in the task list.

from workflows_py.workflow import Step, Workflow

def build_review_tasks(ctx):
    # Your normal step logic here
    return {"ok": True}

def route_to_latest_task(ctx):
    if not ctx.workflow_run_id:
        return None

    # Replace this with the SQL helper your workflow already uses,
    # for example client.v1.sql_execute(...).
    rows = query_tasks(
        """
        select id
        from sample.tasks
        where workflow_run_id = %(workflow_run_id)s
          and status = 'suspended'
        order by created_at desc
        limit 1
        """,
        {"workflow_run_id": ctx.workflow_run_id},
    )

    return rows[0]["id"] if rows else None

workflow = Workflow()
workflow.then(
    Step(
        "build-review-tasks",
        build_review_tasks,
        options={
            "next_task": route_to_latest_task,
        },
    )
)

In practice, query_tasks(...) should wrap whatever SQL access your workflow already uses, such as client.v1.sql_execute(...). The next_task handler receives the normal WorkflowRunContext and should return one of:

A task ID like "tsk_123" to route to that task
None to fall back to Sample’s default next-task routing

next_task is a routing override only. It does not change the workflow graph or which step executes next. The workflow still advances normally. This only controls which task ID is returned to the UI or API after the current step completes.

For “send me to the latest task in this run” behavior, query tasks using ctx.workflow_run_id. The next_task callback is evaluated after the workflow advances, so downstream suspended tasks already exist when this lookup runs.

You can return a task ID from another workflow run. If the task ID is invalid, outside your org, or does not point to a suspended screen task, Sample falls back to the default routing behavior.

Available in workflows-py >= 0.1.28-beta.6.

Combining Control Flow Patterns

You can combine different control flow patterns to create more complex workflows:

from workflows_py.workflow import Step, Workflow

def process_orders(ctx):
    return [
        {"id": 1, "priority": "high", "items": ["A", "B"]},
        {"id": 2, "priority": "low", "items": ["C"]},
        {"id": 3, "priority": "high", "items": ["D", "E", "F"]}
    ]

def process_high_priority(ctx):
    order = ctx.scope["item"]
    return {"order_id": order["id"], "expedited": True}

def process_low_priority(ctx):
    order = ctx.scope["item"]
    return {"order_id": order["id"], "expedited": False}

def send_notification(ctx):
    order = ctx.scope["item"]
    return {"notification_sent": True, "order_id": order["id"]}

workflow = Workflow()
workflow.then(
    Step("get-orders", process_orders)
).loop(
    "process-each-order",
    get_items=lambda ctx: ctx.get_step_result("get-orders"),
    branch_fn=lambda w: w.if_condition(
        condition=lambda ctx: ctx.scope["item"]["priority"] == "high",
        branch_fn=lambda w: w.then(
            Step("high-priority", process_high_priority)
        )
    ).else_branch(
        branch_fn=lambda w: w.then(
            Step("low-priority", process_low_priority)
        )
    ).then(
        Step("notify", send_notification)
    )
)

Mix and match control flow patterns to handle complex business logic.

Workflows

Screens

Integrations

Other guides

Control Flow Patterns

Sequential

Conditionals

Loops

Parallel

Sequential

Conditionals

Loops

Batching Loop Iterations

Parallel

Retries

Error Handling

Next Task Routing

Combining Control Flow Patterns

​Control Flow Patterns

Sequential

Conditionals

Loops

Parallel

​Sequential

​Conditionals

​Loops

​Batching Loop Iterations

​Parallel

​Retries

​Error Handling

​Next Task Routing

​Combining Control Flow Patterns

Control Flow Patterns

Sequential

Conditionals

Loops

Batching Loop Iterations

Parallel

Retries

Error Handling

Next Task Routing

Combining Control Flow Patterns