Creation of DTS example and passing of completionToken

Signed-off-by: Ryan Lettieri <[email protected]>

Author: RyanLettieri

durabletask/worker.py

@@ -143,9 +143,11 @@ def run_loop():
                             request_type = work_item.WhichOneof('request')
                             self._logger.debug(f'Received "{request_type}" work item')
                             if work_item.HasField('orchestratorRequest'):
-                                executor.submit(self._execute_orchestrator, work_item.orchestratorRequest, stub)
+                                executor.submit(self._execute_orchestrator, work_item.orchestratorRequest, stub, work_item.completionToken)
                             elif work_item.HasField('activityRequest'):
-                                executor.submit(self._execute_activity, work_item.activityRequest, stub)
+                                executor.submit(self._execute_activity, work_item.activityRequest, stub, work_item.completionToken)
+                            elif work_item.HasField('healthPing'):
+                                pass # no-op
                             else:
                                 self._logger.warning(f'Unexpected work item type: {request_type}')
 
@@ -184,39 +186,42 @@ def stop(self):
         self._logger.info("Worker shutdown completed")
         self._is_running = False
 
-    def _execute_orchestrator(self, req: pb.OrchestratorRequest, stub: stubs.TaskHubSidecarServiceStub):
+    def _execute_orchestrator(self, req: pb.OrchestratorRequest, stub: stubs.TaskHubSidecarServiceStub, completionToken):
         try:
             executor = _OrchestrationExecutor(self._registry, self._logger)
             result = executor.execute(req.instanceId, req.pastEvents, req.newEvents)
             res = pb.OrchestratorResponse(
                 instanceId=req.instanceId,
                 actions=result.actions,
-                customStatus=pbh.get_string_value(result.encoded_custom_status))
+                customStatus=pbh.get_string_value(result.encoded_custom_status),
+                completionToken=completionToken)
         except Exception as ex:
             self._logger.exception(f"An error occurred while trying to execute instance '{req.instanceId}': {ex}")
             failure_details = pbh.new_failure_details(ex)
             actions = [pbh.new_complete_orchestration_action(-1, pb.ORCHESTRATION_STATUS_FAILED, "", failure_details)]
-            res = pb.OrchestratorResponse(instanceId=req.instanceId, actions=actions)
+            res = pb.OrchestratorResponse(instanceId=req.instanceId, actions=actions, completionToken=completionToken)
 
         try:
             stub.CompleteOrchestratorTask(res)
         except Exception as ex:
             self._logger.exception(f"Failed to deliver orchestrator response for '{req.instanceId}' to sidecar: {ex}")
 
-    def _execute_activity(self, req: pb.ActivityRequest, stub: stubs.TaskHubSidecarServiceStub):
+    def _execute_activity(self, req: pb.ActivityRequest, stub: stubs.TaskHubSidecarServiceStub, completionToken):
         instance_id = req.orchestrationInstance.instanceId
         try:
             executor = _ActivityExecutor(self._registry, self._logger)
             result = executor.execute(instance_id, req.name, req.taskId, req.input.value)
             res = pb.ActivityResponse(
                 instanceId=instance_id,
                 taskId=req.taskId,
-                result=pbh.get_string_value(result))
+                result=pbh.get_string_value(result),
+                completionToken=completionToken)
         except Exception as ex:
             res = pb.ActivityResponse(
                 instanceId=instance_id,
                 taskId=req.taskId,
-                failureDetails=pbh.new_failure_details(ex))
+                failureDetails=pbh.new_failure_details(ex),
+                completionToken=completionToken)
 
         try:
             stub.CompleteActivityTask(res)

examples/README.md

@@ -1,14 +1,39 @@
 # Examples
 
-This directory contains examples of how to author durable orchestrations using the Durable Task Python SDK.
+This directory contains examples of how to author durable orchestrations using the Durable Task Python SDK. There are two backends that are compatible with the Durable Task Python SDK: The Dapr sidecar, and the Durable Task Scheduler (DTS)
 
-## Prerequisites
+## Prerequisites for using Dapr
 
 All the examples assume that you have a Durable Task-compatible sidecar running locally. There are two options for this:
 
 1. Install the latest version of the [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/), which contains and exposes an embedded version of the Durable Task engine. The setup process (which requires Docker) will configure the workflow engine to store state in a local Redis container.
 
-1. Clone and run the [Durable Task Sidecar](https://github.com/microsoft/durabletask-go) project locally (requires Go 1.18 or higher). Orchestration state will be stored in a local sqlite database.
+2. Clone and run the [Durable Task Sidecar](https://github.com/microsoft/durabletask-go) project locally (requires Go 1.18 or higher). Orchestration state will be stored in a local sqlite database.
+
+
+## Prerequisites for using DTS
+
+All the examples assume that you have a Durable Task Scheduler taskhub created.
+
+The simplest way to create a taskhub is by using the az cli commands:
+
+1. Create a scheduler:
+    az durabletask scheduler create --resource-group <testrg> --name <testscheduler> --location <eastus> --ip-allowlist "[0.0.0.0/0]" --sku-capacity 1, --sku-name "Dedicated" --tags "{}"
+
+2. Create your taskhub
+    az durabletask taskhub create --resource-group <testrg> --scheduler-name <testscheduler> --name <testtaskhub>
+
+3. Retrieve the endpoint for the taskhub. This can be done by locating the taskhub in the portal.
+
+4. Set the appropriate environment variables for the TASKHUB and ENDPOINT
+
+```sh
+export TASKHUB=<taskhubname>
+```
+
+```sh
+export ENDPOINT=<taskhubEndpoint>
+```
 
 ## Running the examples
 

examples/dts_activity_sequence.py

@@ -0,0 +1,73 @@
+import os
+from azure.identity import DefaultAzureCredential
+
+"""End-to-end sample that demonstrates how to configure an orchestrator
+that calls an activity function in a sequence and prints the outputs."""
+from durabletask import client, task, worker
+
+
+def hello(ctx: task.ActivityContext, name: str) -> str:
+    """Activity function that returns a greeting"""
+    return f'Hello {name}!'
+
+
+def sequence(ctx: task.OrchestrationContext, _):
+    """Orchestrator function that calls the 'hello' activity function in a sequence"""
+    # call "hello" activity function in a sequence
+    result1 = yield ctx.call_activity(hello, input='Tokyo')
+    result2 = yield ctx.call_activity(hello, input='Seattle')
+    result3 = yield ctx.call_activity(hello, input='London')
+
+    # return an array of results
+    return [result1, result2, result3]
+
+
+# Read the environment variable
+taskhub_name = os.getenv("TASKHUB")
+
+# Check if the variable exists
+if taskhub_name:
+    print(f"The value of TASKHUB is: {taskhub_name}")
+else:
+    print("TASKHUB is not set. Please set the TASKHUB environment variable to the name of the taskhub you wish to use")
+    print("If you are using windows powershell, run the following: $env:TASKHUB=\"<taskhubname>\"")
+    print("If you are using bash, run the following: export TASKHUB=\"<taskhubname>\"")
+    exit()
+
+# Read the environment variable
+endpoint = os.getenv("ENDPOINT")
+
+# Check if the variable exists
+if endpoint:
+    print(f"The value of ENDPOINT is: {endpoint}")
+else:
+    print("ENDPOINT is not set. Please set the ENDPOINT environment variable to the endpoint of the taskhub")
+    print("If you are using windows powershell, run the following: $env:ENDPOINT=\"<taskhubEndpoint>\"")
+    print("If you are using bash, run the following: export ENDPOINT=\"<taskhubEndpoint>\"")
+    exit()
+
+
+default_credential = DefaultAzureCredential()
+# Define the scope for Azure Resource Manager (ARM)
+arm_scope = "https://durabletask.io/.default"
+
+# Retrieve the access token
+access_token = "Bearer " + default_credential.get_token(arm_scope).token
+# create a client, start an orchestration, and wait for it to finish
+metaData: list[tuple[str, str]] = [
+    ("taskhub", taskhub_name), # Hardcode for now, just the taskhub name
+    ("authorization", access_token) # use azure identity sdk for python
+]
+# configure and start the worker
+with worker.TaskHubGrpcWorker(host_address=endpoint, metadata=metaData, secure_channel=True) as w:
+    w.add_orchestrator(sequence)
+    w.add_activity(hello)
+    w.start()
+
+    c = client.TaskHubGrpcClient(host_address=endpoint, metadata=metaData, secure_channel=True)
+    instance_id = c.schedule_new_orchestration(sequence)
+    state = c.wait_for_orchestration_completion(instance_id, timeout=45)
+    if state and state.runtime_status == client.OrchestrationStatus.COMPLETED:
+        print(f'Orchestration completed! Result: {state.serialized_output}')
+    elif state:
+        print(f'Orchestration failed: {state.failure_details}')