Workflows

What is a Workflow?

A workflow is the top-level orchestration function that coordinates multiple steps, handles business logic, and can pause for extended periods using sleep or webhooks. Workflows are the entry point for your business processes.

from pyworkflow import workflow, step, start, sleep

@workflow()
async def order_processing(order_id: str):
    # Validate and process the order
    order = await validate_order(order_id)
    payment = await process_payment(order)

    # Wait for fulfillment
    await sleep("2h")

    # Ship and notify
    await create_shipment(order)
    await send_confirmation(order)

    return {"status": "completed", "order_id": order_id}

Key Characteristics

Durable

Workflows survive crashes, restarts, and deployments. State is preserved through event sourcing.

Suspendable

Workflows can pause for minutes, hours, or days without consuming resources.

Distributed

Workflows execute across Celery workers, enabling horizontal scaling.

Deterministic

Workflows can be replayed from any point using the recorded event log.

Creating Workflows

Decorator
Class

from pyworkflow import workflow

@workflow()
async def my_workflow(user_id: str, amount: float):
    # Your workflow logic here
    result = await some_step(user_id, amount)
    return result

from pyworkflow import Workflow

class MyWorkflow(Workflow):
    async def run(self, user_id: str, amount: float):
        result = await SomeStep()(user_id, amount)
        return result

Configuration Options

Decorator
Class

@workflow(
    name="custom_workflow_name",  # Override the function name
    max_duration="24h",           # Maximum workflow runtime
    max_retries=3                 # Retry the entire workflow on failure
)
async def my_workflow():
    pass

class MyWorkflow(Workflow):
    name = "custom_workflow_name"
    max_duration = "24h"
    max_retries = 3

    async def run(self):
        pass

Option	Type	Default	Description
`name`	`str`	Function/class name	Unique identifier for the workflow
`max_duration`	`str`	`"7d"`	Maximum time the workflow can run
`max_retries`	`int`	`0`	Number of times to retry the entire workflow

Starting Workflows

Synchronous Start

The start() function dispatches a workflow to Celery and returns immediately:

from pyworkflow import start

# Start the workflow (non-blocking)
run_id = start(order_processing, order_id="ORD-123")
print(f"Workflow started with ID: {run_id}")

With Idempotency Key

Prevent duplicate workflow executions:

run_id = start(
    order_processing,
    order_id="ORD-123",
    idempotency_key="order-ORD-123"
)

# Calling again with same key returns the same run_id
run_id_2 = start(
    order_processing,
    order_id="ORD-123",
    idempotency_key="order-ORD-123"
)

assert run_id == run_id_2  # True - same workflow

Workflow Lifecycle

┌─────────────┐
│   PENDING   │  Workflow created, waiting to start
└──────┬──────┘
       │
       ▼
┌─────────────┐
│   RUNNING   │  Workflow is executing
└──────┬──────┘
       │
       ├────────────────┬────────────────┐
       │                │                │
       ▼                ▼                ▼
┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│  SUSPENDED  │  │ INTERRUPTED │  │   FAILED    │
│             │  │(worker crash)│  │             │
└──────┬──────┘  └──────┬──────┘  └─────────────┘
       │                │
       │ (sleep ends)   │ (auto recovery)
       ▼                ▼
┌─────────────┐  ┌─────────────┐
│   RUNNING   │  │   RUNNING   │  Recovered
└──────┬──────┘  └──────┬──────┘
       │                │
       └────────┬───────┘
                │
                ▼
         ┌─────────────┐
         │  COMPLETED  │  Workflow finished successfully
         └─────────────┘

When a worker crashes, the workflow enters INTERRUPTED status and automatically recovers on another worker. See Fault Tolerance for details.

Workflow Context

Inside a workflow, you can access the execution context:

from pyworkflow import workflow, get_context

@workflow()
async def my_workflow():
    ctx = get_context()

    print(f"Run ID: {ctx.run_id}")
    print(f"Workflow: {ctx.workflow_name}")

    # Access step results from replay
    previous_result = ctx.step_results.get("step_id")

Error Handling

Workflows automatically handle errors based on their type:

from pyworkflow import workflow, FatalError, RetryableError

@workflow(max_retries=3)
async def my_workflow():
    try:
        result = await risky_operation()
        return result
    except ValidationError as e:
        # Fatal errors stop the workflow immediately
        raise FatalError(f"Invalid input: {e}")
    except TemporaryError as e:
        # Retryable errors trigger workflow retry
        raise RetryableError(f"Temporary failure: {e}")

Best Practices

Keep workflows focused

Each workflow should handle a single business process. If a workflow is getting complex, consider breaking it into smaller workflows that call each other.

Use meaningful names

Workflow names should clearly describe their purpose: process_order, onboard_user, send_notification_sequence.

Handle idempotency

Use idempotency keys for workflows that shouldn’t run twice for the same input. This prevents duplicate processing during retries.

Set appropriate timeouts

Use max_duration to prevent workflows from running indefinitely. Consider the longest possible execution path.

Configure fault tolerance

Enable recover_on_worker_loss for critical workflows to ensure automatic recovery from worker crashes. See Fault Tolerance for configuration options.

Next Steps

Steps

Learn about steps - the building blocks of workflows.

Step Context

Share typed context data with distributed steps.

Schedules

Automatically run workflows on cron, interval, or calendar schedules.

Fault Tolerance

Configure auto recovery from worker crashes.

Getting Started

Core Concepts

Guides

What is a Workflow?

Key Characteristics

Durable

Suspendable

Distributed

Deterministic

Creating Workflows

Configuration Options

Starting Workflows

Synchronous Start

With Idempotency Key

Workflow Lifecycle

Workflow Context

Error Handling

Best Practices

Next Steps

Steps

Step Context

Schedules

Fault Tolerance

​What is a Workflow?

​Key Characteristics

Durable

Suspendable

Distributed

Deterministic

​Creating Workflows

​Configuration Options

​Starting Workflows

​Synchronous Start

​With Idempotency Key

​Workflow Lifecycle

​Workflow Context

​Error Handling

​Best Practices

​Next Steps

Steps

Step Context

Schedules

Fault Tolerance

What is a Workflow?

Key Characteristics

Creating Workflows

Configuration Options

Starting Workflows

Synchronous Start

With Idempotency Key

Workflow Lifecycle

Workflow Context

Error Handling

Best Practices

Next Steps