workflow

DAPR Workflow Scaffolding

Create a new DAPR workflow with all required components and best practices.

Behavior

When the user runs /dapr:workflow:

1. Gather Workflow Info

   • Workflow name (from $ARGUMENTS or prompt)
   • Workflow pattern (sequential, parallel, saga, approval)
   • Number of activities
   • Brief description of each step

2. Generate Workflow Code

   # workflows/{workflow_name}.py
   from dapr.ext.workflow import (
       WorkflowRuntime,
       DaprWorkflowContext,
       WorkflowActivityContext,
       RetryPolicy
   )
   from datetime import timedelta
   
   wf_runtime = WorkflowRuntime()
   
   # Retry policy
   retry_policy = RetryPolicy(
       max_number_of_attempts=3,
       first_retry_interval=timedelta(seconds=1),
       backoff_coefficient=2.0
   )
   
   @wf_runtime.workflow(name="{workflow_name}")
   def {workflow_name}_workflow(ctx: DaprWorkflowContext, input: dict):
       # Generated workflow logic
       pass
   
   @wf_runtime.activity(name="{activity_name}")
   def {activity_name}(ctx: WorkflowActivityContext, data: dict):
       # Generated activity logic
       pass
   

3. Generate Client Code

   # clients/{workflow_name}_client.py
   from dapr.ext.workflow import DaprWorkflowClient
   
   async def start_{workflow_name}(input_data: dict):
       async with DaprWorkflowClient() as client:
           instance_id = await client.schedule_new_workflow(
               workflow="{workflow_name}",
               input=input_data
           )
           return instance_id
   

4. Generate Tests

   # tests/test_{workflow_name}.py
   import pytest
   from workflows.{workflow_name} import *
   
   def test_{workflow_name}_activity():
       # Test activity logic
       pass
   

5. Update Dependencies

   • Add dapr-ext-workflow to requirements.txt if missing

Arguments

• $ARGUMENTS - Workflow name and pattern:
  • order-processing - Basic workflow name
  • order-processing --pattern saga - With specific pattern
  • order-processing --steps 5 - Number of steps