Merge pull request #1191 from marcduiker/add-workflow-tutorials-python

Workflow tutorials for Python

Author: alicejgibbons

tutorials/workflow/csharp/workflow-management/README.md

@@ -13,7 +13,7 @@ For more information on workflow management, see the [Dapr docs](https://docs.da
 
 ## Inspect the code
 
-Open the `Program.cs` file in the `tutorials/workflow/csharp/child-workflows/WorkflowManagement` folder. This file contains the endpoint definitions that use the workflow management API. The workflow that is being managed is named `NeverEndingWorkflow` and is a counter that will keep running once it's started.
+Open the `Program.cs` file in the `tutorials/workflow/csharp/workflow-management/WorkflowManagement` folder. This file contains the endpoint definitions that use the workflow management API. The workflow that is being managed is named `NeverEndingWorkflow` and is a counter that will keep running once it's started.
 
 ## Run the tutorial
 

tutorials/workflow/python/README.md

@@ -0,0 +1,25 @@
+# Using the Dapr Workflow API with Python
+
+This folder contains tutorials of using the Dapr Workflow API with Python. All examples can be run locally on your machine.
+
+Before you start, it's recommended to read though the Dapr docs to get familiar with the many [Workflow features, concepts, and patterns](https://docs.dapr.io/developing-applications/building-blocks/workflow/).
+
+## Prerequisites
+
+- [Docker Desktop](https://www.docker.com/products/docker-desktop/)
+- [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/) & [Initialization](https://docs.dapr.io/getting-started/install-dapr-selfhost/)
+- [Python 3](https://www.python.org/downloads/)
+- Optional: An IDE such as [VSCode](https://code.visualstudio.com/download) with a [REST client](https://marketplace.visualstudio.com/items?itemName=humao.rest-client).
+
+## Tutorials
+
+- [Workflow Basics](./fundamentals/README.md)
+- [Task Chaining](./task-chaining/README.md)
+- [Fan-out/Fan-in](./fan-out-fan-in/README.md)
+- [Monitor](./monitor-pattern/README.md)
+- [External Events](./external-system-interaction/README.md)
+- [Child Workflows](./child-workflows/README.md)
+- [Resiliency & Compensation](./resiliency-and-compensation/README.md)
+- [Combined Patterns](./combined-patterns/README.md)
+- [WorkflowManagement](./workflow-management/README.md)
+- [Challenges & Tips](./challenges-tips/README.md)

tutorials/workflow/python/challenges-tips/README.md

@@ -0,0 +1,10 @@
+# Workflow Challenges & Tips
+
+Workflow systems are very powerful tools but also have their challenges & limitations as described in the [Dapr docs](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-features-concepts/#limitations).
+
+This section provides some tips with code snippets to understand the limitations and get the most out of the Dapr Workflow API. Read through the following examples to learn best practices to develop Dapr workflows.
+
+- [Deterministic workflows](deterministic_workflow.py)
+- [Idempotent activities](idempotent_activity.py)
+- [Versioning workflows](versioning_workflow.py)
+- [Workflow & activity payload size](payload_size_workflow.py)

tutorials/workflow/python/challenges-tips/deterministic_workflow.py

@@ -0,0 +1,36 @@
+import dapr.ext.workflow as wf
+import datetime
+import uuid
+
+wf_runtime = wf.WorkflowRuntime()
+
+@wf_runtime.workflow(name='non_deterministic_workflow')
+def non_deterministic_workflow(ctx: wf.DaprWorkflowContext, wf_input: str):
+    
+    """
+    Do not use non-deterministic operations in a workflow!
+    These operations will create a new value every time the
+    workflow is replayed.
+    """
+    order_id = str(uuid.uuid4())
+    order_date = datetime.now()
+    yield ctx.call_activity(submit_id, input=order_id)
+    yield ctx.call_activity(submit_date, input=order_date)
+    
+    return order_id
+
+
+@wf_runtime.workflow(name='deterministic_workflow')
+def deterministic_workflow(ctx: wf.DaprWorkflowContext, wf_input: str):
+    
+    """
+    Either wrap non-deterministic operations in an activity. Or use deterministic
+    alternatives on the DaprWorkflowContext instead. These operations create the 
+    same value when the workflow is replayed.
+    """
+    order_id = yield ctx.call_activity(create_order_id, input=wf_input)
+    order_date = ctx.current_utc_datetime
+    yield ctx.call_activity(submit_id, input=order_id)
+    yield ctx.call_activity(submit_date, input=order_date)
+
+    return order_id

tutorials/workflow/python/challenges-tips/idempotent_activity.py

@@ -0,0 +1,14 @@
+import dapr.ext.workflow as wf
+
+wf_runtime = wf.WorkflowRuntime()
+
+@wf_runtime.activity(name='idempotent_activity')
+def idempotent_activity(ctx: wf.WorkflowActivityContext, order_item: Order) -> bool:
+    """
+    Beware of non-idempotent operations in an activity.
+    Dapr Workflow guarantees at-least-once execution of activities, so activities might be executed more than once
+    in case an activity is not ran to completion successfully.
+    For instance, can you insert a record to a database twice without side effects?
+         var insertSql = $"INSERT INTO Orders (Id, Description, UnitPrice, Quantity) VALUES ('{order_item.id}', '{order_item.description}', {order_item.unit_price}, {order_item.quantity})";
+    It's best to perform a check if an record already exists before inserting it.
+    """

tutorials/workflow/python/challenges-tips/payload_size_workflow.py

@@ -0,0 +1,29 @@
+import dapr.ext.workflow as wf
+
+wf_runtime = wf.WorkflowRuntime()
+
+@wf_runtime.workflow(name='large_payload_size_workflow')
+def large_payload_size_workflow(ctx: wf.DaprWorkflowContext, doc_id: str):
+    """
+    Do not pass large payloads between activities.
+    They are stored in the Dapr state store twice, one as output argument 
+    for GetDocument, and once as input argument for UpdateDocument.
+    """
+    document = yield ctx.call_activity(get_document, input=doc_id)
+    updated_document = yield ctx.call_activity(update_document, input=document)
+    
+    # More activities to process the updated document
+
+    return updated_document
+
+@wf_runtime.workflow(name='small_payload_size_workflow')
+def small_payload_size_workflow(ctx: wf.DaprWorkflowContext, doc_id: str):
+    """
+    Do pass small payloads between activities, preferably IDs only, or objects that are quick to (de)serialize in large volumes.
+    Combine multiple actions, such as document retrieval and update, into a single activity, or use the Dapr State Store API to store more data.
+    """
+    updated_doc_id = yield ctx.call_activity(get_and_update_document, input=doc_id)
+    
+    # More activities to process the updated document
+
+    return updated_doc_id

tutorials/workflow/python/challenges-tips/versioning_workflow.py

@@ -0,0 +1,64 @@
+import dapr.ext.workflow as wf
+
+wf_runtime = wf.WorkflowRuntime()
+
+"""
+This is the initial version of the workflow.
+Note that the input argument for both activities is the orderItem (string).
+"""
+@wf_runtime.workflow(name='versioning_workflow_1')
+def versioning_workflow_1(ctx: wf.DaprWorkflowContext, order_item: str):
+    result_a = yield ctx.call_activity(activity_a, input=order_item)
+    result_b = yield ctx.call_activity(activity_b, input=order_item)
+    
+    return result_a + result_b
+
+@wf_runtime.activity(name='activity_a')
+def activity_a(ctx: wf.WorkflowActivityContext, order_item: str) -> int:
+    """
+    This activity processes the order item and returns an integer result.
+    """
+    print(f'activity_a: Received input: {order_item}.', flush=True)
+    return 10
+
+@wf_runtime.activity(name='activity_b')
+def activity_b(ctx: wf.WorkflowActivityContext, order_item: str) -> int:
+    """
+    This activity processes the order item and returns another integer result.
+    """
+    print(f'activity_b: Received input: {order_item}.', flush=True)
+    return 20
+
+"""
+This is the updated version of the workflow.
+The input for activity_b has changed from order_item (string) to result_a (int).
+If there are in-flight workflow instances that were started with the previous version
+of this workflow, these will fail when the new version of the workflow is deployed 
+and the workflow name remains the same, since the runtime parameters do not match with the persisted state.
+It is recommended to version workflows by creating a new workflow class with a new name:
+{workflowname}_1 -> {workflowname}_2
+Try to avoid making breaking changes in perpetual workflows (that use the `continue_as_new` method) 
+since these are difficult to replace with a new version.
+"""
+@wf_runtime.workflow(name='versioning_workflow_2')
+def versioning_workflow_2(ctx: wf.DaprWorkflowContext, order_item: str):
+    result_a = yield ctx.call_activity(activity_a, input=order_item)
+    result_b = yield ctx.call_activity(activity_b, input=result_a)
+    
+    return result_a + result_b
+
+@wf_runtime.activity(name='activity_a')
+def activity_a(ctx: wf.WorkflowActivityContext, order_item: str) -> int:
+    """
+    This activity processes the order item and returns an integer result.
+    """
+    print(f'activity_a: Received input: {order_item}.', flush=True)
+    return 10
+
+@wf_runtime.activity(name='activity_b')
+def activity_b(ctx: wf.WorkflowActivityContext, number: int) -> int:
+    """
+    This activity processes a number and returns another integer result.
+    """
+    print(f'activity_b: Received input: {number}.', flush=True)
+    return number + 10

tutorials/workflow/python/child-workflows/README.md

@@ -0,0 +1,112 @@
+# Child Workflows
+
+This tutorial demonstrates how a workflow can call child workflows that are part of the same application. Child workflows can be used to break up large workflows into smaller, reusable parts. For more information about child workflows see the [Dapr docs](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-features-concepts/#child-workflows).
+
+## Inspect the code
+
+Open the `parent_child_workflow.py` file in the `tutorials/workflow/python/child-workflows/child_workflows` folder. This file contains the definition for the workflows and activities.
+
+The parent workflow iterates over the input array and schedules an instance of the `child_workflow` for each of the input elements. The `child_workflow` is a basic task-chaining workflow that contains a sequence of two activities. When all of the instances of the `child_workflow` complete, then the `parent_workflow` finishes.
+
+### Parent workflow
+
+```mermaid
+graph LR
+   SW((Start
+   Workflow))
+   subgraph for each word in the input
+    GWL[Call child workflow]
+   end
+   ALL[Wait until all tasks
+   are completed]
+   EW((End
+   Workflow))
+   SW --> GWL
+   GWL --> ALL
+   ALL --> EW
+```
+
+### Child workflow
+
+```mermaid
+graph LR
+   SW((Start
+   Workflow))
+   A1[activity1]
+   A2[activity2]
+   EW((End
+   Workflow))
+   SW --> A1
+   A1 --> A2
+   A2 --> EW
+```
+
+## Run the tutorial
+
+1. Use a terminal to navigate to the `tutorials/workflow/python/child-workflows/child_workflows` folder.
+2. Install the dependencies using pip:
+
+    ```bash
+    pip3 install -r requirements.txt
+    ```
+
+3. Navigate one level back to the `child-workflows` folder and use the Dapr CLI to run the Dapr Multi-App run file
+
+    <!-- STEP
+    name: Run multi app run template
+    expected_stdout_lines:
+    - 'Started Dapr with app id "childworkflows"'
+    expected_stderr_lines:
+    working_dir: .
+    output_match_mode: substring
+    background: true
+    sleep: 15
+    timeout_seconds: 30
+    -->
+    ```bash
+    dapr run -f .
+    ```
+    <!-- END_STEP -->
+
+4. Use the POST request in the [`childworkflows.http`](./childworkflows.http) file to start the workflow, or use this cURL command:
+
+    ```bash
+    curl -i --request POST \
+    --url http://localhost:5259/start \
+    --header 'content-type: application/json' \
+    --data '["Item 1","Item 2"]'
+    ```
+
+    The input of the workflow is an array with two strings:
+
+    ```json
+    [
+        "Item 1",
+        "Item 2"
+    ]
+    ```
+
+    The app logs should show both the items in the input values array being processed by each activity in the child workflow as follows:
+
+    ```text
+    == APP - childworkflows == activity1: Received input: Item 1.
+    == APP - childworkflows == activity2: Received input: Item 1 is processed.
+    == APP - childworkflows == activity1: Received input: Item 2.
+    == APP - childworkflows == activity2: Received input: Item 2 is processed.
+    ```
+
+5. Use the GET request in the [`childworkflows.http`](./childworkflows.http) file to get the status of the workflow, or use this cURL command:
+
+    ```bash
+    curl --request GET --url http://localhost:3559/v1.0/workflows/dapr/<INSTANCEID>
+    ```
+
+    Where `<INSTANCEID>` is the workflow instance ID you received in the `instance_id` property in the previous step.
+
+    The expected serialized output of the workflow is an array with two strings:
+
+    ```txt
+    "[\"Item 1 is processed as a child workflow.\",\"Item 2 is processed as a child workflow.\"]"
+    ```
+
+6. Stop the Dapr Multi-App run process by pressing `Ctrl+C`.

tutorials/workflow/python/child-workflows/child_workflows/app.py

@@ -0,0 +1,26 @@
+from fastapi import FastAPI, status
+from contextlib import asynccontextmanager
+from typing import List
+from parent_child_workflow import wf_runtime, parent_workflow
+import dapr.ext.workflow as wf
+import uvicorn
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    wf_runtime.start()
+    yield
+    wf_runtime.shutdown()
+
+app = FastAPI(lifespan=lifespan)
+
+@app.post("/start", status_code=status.HTTP_202_ACCEPTED)
+async def start_workflow(items: List[str]):
+    wf_client = wf.DaprWorkflowClient()
+    instance_id = wf_client.schedule_new_workflow(
+            workflow=parent_workflow,
+            input=items
+        )
+    return {"instance_id": instance_id}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=5259)

tutorials/workflow/python/child-workflows/child_workflows/parent_child_workflow.py

@@ -0,0 +1,30 @@
+from typing import List
+import dapr.ext.workflow as wf
+
+wf_runtime = wf.WorkflowRuntime()
+
+@wf_runtime.workflow(name='parent_workflow')
+def parent_workflow(ctx: wf.DaprWorkflowContext, items: List[str]):
+    
+    child_wf_tasks = [
+        ctx.call_child_workflow(child_workflow, input=item) for item in items
+    ]
+    wf_result = yield wf.when_all(child_wf_tasks)
+
+    return wf_result
+
+@wf_runtime.workflow(name='child_workflow')
+def child_workflow(ctx: wf.DaprWorkflowContext, wf_input: str):
+    result1 = yield ctx.call_activity(activity1, input=wf_input)
+    wf_result = yield ctx.call_activity(activity2, input=result1)
+    return wf_result
+
+@wf_runtime.activity(name='activity1')
+def activity1(ctx: wf.WorkflowActivityContext, act_input: str) -> str:
+    print(f'activity1: Received input: {act_input}.', flush=True)
+    return f"{act_input} is processed"
+
+@wf_runtime.activity(name='activity2')
+def activity2(ctx: wf.WorkflowActivityContext, act_input: str) -> str:
+    print(f'activity2: Received input: {act_input}.', flush=True)
+    return f"{act_input} as a child workflow."