updates, test examples, cleanup

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

examples/workflow-async/README.md

@@ -3,11 +3,36 @@
 These examples mirror `examples/workflow/` but author orchestrators with `async def` using the
 async workflow APIs. Activities remain regular functions unless noted.
 
+## Prerequisites
+
+- [Dapr CLI and initialized environment](https://docs.dapr.io/getting-started)
+- [Install Python 3.10+](https://www.python.org/downloads/)
+
+
 How to run:
-- Ensure a Dapr sidecar is running locally. If needed, set `DURABLETASK_GRPC_ENDPOINT`, or
-  `DURABLETASK_GRPC_HOST/PORT`.
-- Install requirements: `pip install -r requirements.txt`
-- Run any example: `python simple.py`
+- Install Dapr CLI: `brew install dapr/tap/dapr-cli` or `choco install dapr-cli`
+- Initialize Dapr: `dapr init`
+- Install requirements: 
+  ```bash
+  cd examples/workflow-async
+  python -m venv .venv
+  source .venv/bin/activate
+  pip install -r requirements.txt
+  ```
+  
+  or better yet with faster `uv`:
+  ```bash
+  uv venv .venv
+  source .venv/bin/activate
+  uv pip install -r requirements.txt
+  ```
+- Run any example with dapr:
+  - `dapr run --app-id wf_async_symple -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python simple.py`
+  - `dapr run --app-id wf_task_chain -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python task_chaining.py`
+  - `dapr run --app-id wf_async_child -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python child_workflow.py`
+  - `dapr run --app-id wf_async_fafi -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python fan_out_fan_in.py`
+  - `dapr run --app-id wf_async_gather -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python fan_out_fan_in_with_gather.py`
+  - `dapr run --app-id wf_async_approval -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python human_approval.py`
 
 Notes:
 - Orchestrators use `await ctx.activity(...)`, `await ctx.sleep(...)`, `await ctx.when_all/when_any(...)`, etc.

examples/workflow-async/child_workflow.py

@@ -17,6 +17,7 @@
     AsyncWorkflowContext,
     DaprWorkflowClient,
     WorkflowRuntime,
+    WorkflowStatus,
 )
 
 wfr = WorkflowRuntime()
@@ -35,12 +36,20 @@ async def parent(ctx: AsyncWorkflowContext, n: int) -> int:
 
 
 def main():
-    wfr.start()
-    client = DaprWorkflowClient()
-    instance_id = 'parent_async_instance'
-    client.schedule_new_workflow(workflow=parent, input=5, instance_id=instance_id)
-    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
-    wfr.shutdown()
+    with wfr:
+        # the context manager starts the workflow runtime on __enter__ and shutdown on __exit__
+        client = DaprWorkflowClient()
+        instance_id = 'parent_async_instance'
+        client.schedule_new_workflow(workflow=parent, input=5, instance_id=instance_id)
+        wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+
+    # simple test
+    if wf_state.runtime_status != WorkflowStatus.COMPLETED:
+        print('Workflow failed with status ', wf_state.runtime_status)
+        exit(1)
+    if wf_state.serialized_output != '11':
+        print('Workflow result is incorrect!')
+        exit(1)
 
 
 if __name__ == '__main__':

examples/workflow-async/fan_out_fan_in.py

@@ -17,6 +17,7 @@
     DaprWorkflowClient,
     WorkflowActivityContext,
     WorkflowRuntime,
+    WorkflowStatus,
 )
 
 wfr = WorkflowRuntime()
@@ -30,6 +31,7 @@ def square(ctx: WorkflowActivityContext, x: int) -> int:
 @wfr.async_workflow(name='fan_out_fan_in_async')
 async def orchestrator(ctx: AsyncWorkflowContext):
     tasks = [ctx.call_activity(square, input=i) for i in range(1, 6)]
+    # 1 + 4 + 9 + 16 + 25 = 55
     results = await ctx.when_all(tasks)
     total = sum(results)
     return total
@@ -44,6 +46,14 @@ def main():
     print(f'Workflow state: {wf_state}')
     wfr.shutdown()
 
+    # simple test
+    if wf_state.runtime_status != WorkflowStatus.COMPLETED:
+        print('Workflow failed with status ', wf_state.runtime_status)
+        exit(1)
+    if wf_state.serialized_output != '55':
+        print('Workflow result is incorrect!')
+        exit(1)
+
 
 if __name__ == '__main__':
     main()

examples/workflow-async/fan_out_fan_in_with_gather.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+
+import asyncio
+
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+    WorkflowStatus,
+)
+
+# test using sandbox to convert asyncio methods into deterministic ones
+
+wfr = WorkflowRuntime()
+
+
+@wfr.activity(name='square')
+def square(ctx: WorkflowActivityContext, x: int) -> int:
+    return x * x
+
+
+#  workflow function auto-recognize coroutine function and converts this into wfr.async_workflow
+@wfr.workflow(name='fan_out_fan_in_async')
+async def orchestrator(ctx: AsyncWorkflowContext):
+    tasks = [ctx.call_activity(square, input=i) for i in range(1, 6)]
+    # 1 + 4 + 9 + 16 + 25 = 55
+    results = await asyncio.gather(*tasks)
+    total = sum(results)
+    return total
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'fofi_async'
+    client.schedule_new_workflow(workflow=orchestrator, instance_id=instance_id)
+    wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    print(f'Workflow state: {wf_state}')
+    wfr.shutdown()
+
+    # simple test
+    if wf_state.runtime_status != WorkflowStatus.COMPLETED:
+        print('Workflow failed with status ', wf_state.runtime_status)
+        exit(1)
+    if wf_state.serialized_output != '55':
+        print('Workflow result is incorrect!')
+        exit(1)
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/human_approval.py

@@ -13,23 +13,35 @@
 limitations under the License.
 """
 
-from dapr.ext.workflow import AsyncWorkflowContext, DaprWorkflowClient, WorkflowRuntime
+import time
+from datetime import timedelta
+
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowRuntime,
+    WorkflowStatus,
+)
 
 wfr = WorkflowRuntime()
 
 
 @wfr.async_workflow(name='human_approval_async')
 async def orchestrator(ctx: AsyncWorkflowContext, request_id: str):
+    approve = ctx.wait_for_external_event(f'approve:{request_id}')
+    reject = ctx.wait_for_external_event(f'reject:{request_id}')
     decision = await ctx.when_any(
         [
-            ctx.wait_for_external_event(f'approve:{request_id}'),
-            ctx.wait_for_external_event(f'reject:{request_id}'),
-            ctx.create_timer(300.0),
+            approve,
+            reject,
+            ctx.create_timer(timedelta(seconds=5)),
         ]
     )
-    if isinstance(decision, dict) and decision.get('approved'):
-        return 'APPROVED'
-    if isinstance(decision, dict) and decision.get('rejected'):
+    if decision == approve:
+        print(f'Decision Approved')
+        return request_id
+    if decision == reject:
+        print(f'Decision Rejected')
         return 'REJECTED'
     return 'TIMEOUT'
 
@@ -38,10 +50,29 @@ def main():
     wfr.start()
     client = DaprWorkflowClient()
     instance_id = 'human_approval_async_1'
-    client.schedule_new_workflow(workflow=orchestrator, input='REQ-1', instance_id=instance_id)
+    try:
+        # clean up previous workflow with this ID
+        client.terminate_workflow(instance_id)
+        client.purge_workflow(instance_id)
+    except Exception:
+        pass
+    client.schedule_new_workflow(workflow=orchestrator, input='req-1', instance_id=instance_id)
+    time.sleep(1)
+    client.raise_workflow_event(instance_id, 'approve:req-1')
     # In a real scenario, raise approve/reject event from another service.
+    wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=20)
+    print(f'Workflow state: {wf_state}')
+
     wfr.shutdown()
 
+    # simple test
+    if wf_state.runtime_status != WorkflowStatus.COMPLETED:
+        print('Workflow failed with status ', wf_state.runtime_status)
+        exit(1)
+    if wf_state.serialized_output != '"req-1"':
+        print('Workflow result is incorrect!')
+        exit(1)
+
 
 if __name__ == '__main__':
     main()

examples/workflow-async/requirements.txt

@@ -1,2 +1,13 @@
-dapr-ext-workflow-dev>=1.15.0.dev
-dapr-dev>=1.15.0.dev
+#dapr-ext-workflow-dev>=1.15.0.dev
+#dapr-dev>=1.15.0.dev
+
+
+## --- local development: install local packages in editable mode --
+
+## -- if using dev version of durabletask-python
+-e ../../../durabletask-python
+
+## -- if using dev version of dapr-ext-workflow
+-e ../..
+-e ../../ext/dapr-ext-workflow
+

examples/workflow-async/simple.py

@@ -11,6 +11,7 @@
 See the specific language governing permissions and
 limitations under the License.
 """
+import json
 
 from datetime import timedelta
 from time import sleep
@@ -21,6 +22,7 @@
     RetryPolicy,
     WorkflowActivityContext,
     WorkflowRuntime,
+    WorkflowStatus,
 )
 
 counter = 0
@@ -58,18 +60,19 @@ async def hello_world_wf(ctx: AsyncWorkflowContext, wf_input):
     print(f'Child workflow returned {result_4}')
 
     # Event vs timeout using when_any
+    event_1 = ctx.wait_for_external_event(event_name)
     first = await ctx.when_any(
         [
-            ctx.wait_for_external_event(event_name),
+            event_1,
             ctx.create_timer(timedelta(seconds=30)),
         ]
     )
 
     # Proceed only if event won
-    if isinstance(first, dict) and 'event' in first:
-        await ctx.call_activity(hello_act, input=100)
-        await ctx.call_activity(hello_act, input=1000)
-        return 'Completed'
+    if first == event_1:
+        result_5 = await ctx.call_activity(hello_act, input=100)
+        result_6 = await ctx.call_activity(hello_act, input=1000)
+        return dict(result_1=result_1, result_2=result_2, result_3=result_3, result_4=result_4, result_5=result_5, result_6=result_6)
     return 'Timeout'
 
 
@@ -108,28 +111,37 @@ def act_for_child_wf(ctx: WorkflowActivityContext, inp):
 
 
 def main():
-    wfr.start()
-    wf_client = DaprWorkflowClient()
-
-    wf_client.schedule_new_workflow(
-        workflow=hello_world_wf, input=input_data, instance_id=instance_id
-    )
-
-    wf_client.wait_for_workflow_start(instance_id)
-
-    # Let initial activities run
-    sleep(5)
-
-    # Raise event to continue
-    wf_client.raise_workflow_event(
-        instance_id=instance_id, event_name=event_name, data={'ok': True}
-    )
-
-    # Wait for completion
-    state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
-    print(f'Workflow status: {state.runtime_status.name}')
-
-    wfr.shutdown()
+    wf_state = {}
+    with wfr:
+        wf_client = DaprWorkflowClient()
+
+        wf_client.schedule_new_workflow(
+            workflow=hello_world_wf, input=input_data, instance_id=instance_id
+        )
+
+        wf_client.wait_for_workflow_start(instance_id)
+
+        # Let initial activities run
+        sleep(5)
+
+        # Raise event to continue
+        wf_client.raise_workflow_event(
+            instance_id=instance_id, event_name=event_name, data={'ok': True}
+        )
+
+        # Wait for completion
+        wf_state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+
+    # simple test
+    if wf_state.runtime_status != WorkflowStatus.COMPLETED:
+        print('Workflow failed with status ', wf_state.runtime_status)
+        exit(1)
+    output = json.loads(wf_state.serialized_output)
+    if (output["result_1"] != 'Activity returned 1' or output["result_2"] != 'Activity returned 10' or
+            output["result_3"] != 'Activity returned 2' or output["result_4"] != 'ok' or
+            output["result_5"] != 'Activity returned 100' or output["result_6"] != 'Activity returned 1000'):
+        print('Workflow result is incorrect!')
+        exit(1)
 
 
 if __name__ == '__main__':

examples/workflow-async/task_chaining.py

@@ -17,6 +17,7 @@
     DaprWorkflowClient,
     WorkflowActivityContext,
     WorkflowRuntime,
+    WorkflowStatus,
 )
 
 wfr = WorkflowRuntime()
@@ -40,9 +41,18 @@ def main():
     client = DaprWorkflowClient()
     instance_id = 'task_chain_async'
     client.schedule_new_workflow(workflow=orchestrator, instance_id=instance_id)
-    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
     wfr.shutdown()
 
+    # simple test
+    if wf_state.runtime_status != WorkflowStatus.COMPLETED:
+        print('Workflow failed with status ', wf_state.runtime_status)
+        exit(1)
+    # 1 + 2 + 3 + 4 = 10
+    if wf_state.serialized_output != '10':
+        print('Workflow result is incorrect!')
+        exit(1)
+
 
 if __name__ == '__main__':
     main()