> ## Documentation Index
> Fetch the complete documentation index at: https://docs.samplehc.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Control Flow

> Learn how to structure workflow execution

## Control Flow Patterns

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

<CardGroup cols={2}>
  <Card title="Sequential" icon="arrow-right" href="#sequential">
    Execute steps one after another in a defined order
  </Card>

  <Card title="Conditionals" icon="split" href="#conditionals">
    Branch execution based on runtime conditions
  </Card>

  <Card title="Loops" icon="repeat" href="#loops">
    Process collections of items with repeated logic
  </Card>

  <Card title="Parallel" icon="square-stack" href="#parallel">
    Execute multiple branches simultaneously
  </Card>
</CardGroup>

## Sequential

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

```python theme={null}
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)
)
```

<Tip>
  Each step can access the results of previous steps using
  `ctx.get_step_result("step-id")`.
</Tip>

## Conditionals

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

<CodeGroup>
  ```python Simple If/Else theme={null}
  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)
      )
  )
  ```

  ```python Multiple Conditions theme={null}
  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))
  )
  ```

  ```python Using the result from a conditional theme={null}
  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)
  )
  ```
</CodeGroup>

<Warning>
  Make sure your condition functions handle missing or invalid data gracefully.
</Warning>

## 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).

<CodeGroup>
  ```python Simple List Processing theme={null}
  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)
      )
  )
  ```

  ```python Processing Objects theme={null}
  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)
      )
  )
  ```
</CodeGroup>

<Info>Loop iterations run in parallel by default.</Info>

### 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](/workflows/suspend-resume#delays-and-scheduled-resumes) based on
its index.

```python theme={null}
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)
    )
)
```

<Tip>
  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](/workflows/suspend-resume#delays-and-scheduled-resumes) for how the
  sleep pattern works.
</Tip>

## Parallel

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

```python theme={null}
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)
)
```

<Check>
  Parallel branches are perfect for independent tasks that can run
  simultaneously.
</Check>

## Retries

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

```python theme={null}
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`.

<Note>Available in `workflows-py >= 0.1.21`.</Note>

## 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.

```python theme={null}
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)

<Tip>
  Combine `on_failure` with `num_retries` to first attempt automatic recovery,
  then execute custom logic if all retries fail.
</Tip>

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

<Note>Available in `workflows-py >= 0.1.28`.</Note>

## 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.

```python theme={null}
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

<Tip>
  `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.
</Tip>

<Tip>
  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.
</Tip>

<Warning>
  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.
</Warning>

<Note>Available in `workflows-py >= 0.1.28-beta.6`.</Note>

## Combining Control Flow Patterns

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

```python theme={null}
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)
    )
)
```

<Tip>Mix and match control flow patterns to handle complex business logic.</Tip>