Skip to main content

​
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)
)
)
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)
)
)
Loop iterations run in parallel by default.

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

​
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.
⌘I