Sample: Order processing workflow in python (#170)

* Workflow order processing sample in Python

Signed-off-by: Chris Gillum <cgillum@microsoft.com>

* Update README.md with link to new sample

Signed-off-by: Chris Gillum <cgillum@microsoft.com>

---------

Signed-off-by: Chris Gillum <cgillum@microsoft.com>
This commit is contained in:
Chris Gillum 2023-11-13 15:38:49 -08:00 committed by GitHub
parent 5228153f59
commit fe2e0c109a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 1127 additions and 0 deletions

View File

@ -33,6 +33,7 @@ If you are new to Dapr, you may want to review following resources first:
| [commercetools GraphQL sample output binding](./commercetools-graphql-sample/) | Connects to commercetools, allowing you to query or manipulate a commercetools projects using a provided GraphlQL query. |
| [WebAssembly Middleware](./hello-wasm) | Demonstrates how to serve HTTP responses directly from the dapr sidecar using WebAssembly. |
| [Workflow + external endpoint invocation](./workflow-external-invocation) | Demonstrates how to use the Dapr Workflow API to coordinate an order process that includes an activity which uses service invocation for non-Dapr endpoints. |
| [Workflow + multi-app microservice in Python](./workflow-orderprocessing-python) | Demonstrates how to use the Dapr Workflow Python SDK to coordinate an order process across multiple dapr-enabled microservices. |
## External samples

View File

@ -0,0 +1,15 @@
{
"[python]": {
"editor.defaultFormatter": "ms-python.autopep8",
"editor.formatOnSave": true,
"editor.codeActionsOnSave": {
"source.organizeImports": true,
},
"editor.rulers": [
119
],
},
"autopep8.args": [
"--max-line-length=119"
]
}

View File

@ -0,0 +1,254 @@
# Dapr Workflow Order Processing Microservice Sample
## Sample info
| Attribute | Details |
|--------|--------|
| Dapr runtime version | 1.12.0 |
| Dapr Workflow Python SDK | v0.2.0 |
| Language | Python |
| Environment | Local |
An example of using [Dapr Workflows](https://docs.dapr.io/developing-applications/building-blocks/workflow/) to build a microservice architecture for order processing using Python. This example uses pub/sub and service invocation building blocks.
## Overview
This sample is comprised of five microservices:
- **[orderprocessing](./services/orderprocessing/)**: The main entrypoint for the sample. This service receives an order request and starts a new Dapr Workflow instance to process the order.
- **[notifications](./services/notifications/)**: A service receives pub/sub messages from the orderprocessing service and displays notifications to the user.
- **[inventory](./services/inventory/)**: A service for managing inventory that is invoked by the orderprocessing workflows.
- **[payments](./services/payments/)**: A service for managing payments that is invoked by the orderprocessing workflows.
- **[shipping](./services/shipping/)**: A service for managing shipping that is invoked by the orderprocessing workflows.
The order processing workflow and all services are written in Python. Orders are submitted through a Flask REST API in the orderprocessing service, which triggers a workflow instance to process the order. As the workflow progresses, it uses pub/sub to send notifications to the notifications service, and service invocation to invoke the inventory, payments, and shipping services.
Orders that are $1,000 or more must be approved by a designated approver within 24 hours, otherwise the order will be cancelled. The workflow will automatically pause execution to wait for the approval using the [external workflow event](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-features-concepts/#external-events) APIs. Approvals can be submitted through the Flask REST API in the orderprocessing service.
## Prerequisites
1. Python 3.8+
1. [Dapr CLI](https://docs.dapr.io/getting-started/install-dapr-cli/)
- Ensure that you're using **v1.12** of the Dapr runtime and the CLI.
1. A REST client, such as [cURL](https://curl.se/), or the VSCode [REST client extension](https://marketplace.visualstudio.com/items?itemName=humao.rest-client) (recommended).
## Setting up the Python environment
It's recommended to create a virtual environment for the sample. This can be done using the `venv` module in Python 3.8+. You can then install the dependencies for each of the microservices.
1. Create a virtual environment for the sample:
```sh
python3 -m venv .venv
```
1. Activate the virtual environment:
```sh
source .venv/bin/activate
```
1. Install the dependencies for each of the microservices under the `services` directory:
```sh
pip install -r services/orderprocessing/requirements.txt
pip install -r services/notifications/requirements.txt
pip install -r services/inventory/requirements.txt
pip install -r services/payments/requirements.txt
pip install -r services/shipping/requirements.txt
```
## Running the sample
The full microservice architecture can be run locally using the [Multi-App Run](https://docs.dapr.io/developing-applications/local-development/multi-app-dapr-run/) feature of the Dapr CLI. The multi-app configuration is defined in the [dapr.yaml](./dapr.yaml) file.
1. To start all the services, navigate to the root directory of the sample and run the following command:
```sh
dapr run -f .
```
1. Next, navigate to the notification service web UI at http://localhost:3001 to view the notifications from workflows. Keep this page open while running the sample.
1. Submit an order using the following cURL command, or via the [demo.http](./demo.http) file if you are using VSCode with the REST client, to start a new order processing workflow that requires approval (if using PowerShell, replace `\` with `` ` ``):
```sh
curl -i -X POST http://localhost:3000/orders \
-H "Content-Type: application/json" \
-d '{"customer": "Riley", "items": ["milk", "bread", "iPhone"], "total": 1299.00}'
```
Expected result (abbreviated):
```http
HTTP/1.1 202 ACCEPTED
Location: http://localhost:3000/orders/order_riley_o066i
Content-Type: text/html; charset=utf-8
Content-Length: 40
Order received. ID = 'order_riley_o066i'
```
> **NOTE**: `order_riley_o066i` in the response is the order ID. The exact value is random and will be different for each order. In each subsequent example, replace `order_riley_o066i` with the order ID returned from the previous request.
> **NOTE**: If you want to submit an order the is auto-approved, simply change the `total` value to be less than `1000`.
1. Check the status of the order by sending a GET request to the URL in the `Location` header of the previous response:
```http
curl -i http://localhost:3000/orders/order_riley_o066i
```
Expected result (formatted for readability):
```json
{
"created_time": "2023-10-28T17:02:05.137170",
"details": {
"customer": "Riley",
"id": "order_riley_o066i",
"items": ["milk", "bread", "iPhone"],
"total": 1299.0
},
"id": "order_riley_o066i",
"last_updated_time": "2023-10-28T17:02:05.222644",
"status": "RUNNING"
}
```
1. Check the notifications web UI at http://localhost:3001 to see the notifications from the workflow. They should look something like this:
```text
2023-10-28T10:02:05-07:00 | order_riley_o066i | Received order for Riley: ['milk', 'bread', 'iPhone']. Total = 1299.0
2023-10-28T10:02:05-07:00 | order_riley_o066i | Reserved inventory for order: ['milk', 'bread', 'iPhone']
2023-10-28T10:02:05-07:00 | order_riley_o066i | Waiting for approval since order >= 1000.0. Deadline = 2023-10-29 17:02:05.
```
1. Approve the order by sending a POST request to the approval endpoint:
```sh
curl -i -X POST http://localhost:3000/orders/order_riley_o066i/approve \
-H "Content-Type: application/json" \
-d '{"approver": "Chris", "approved": true}'
```
> **NOTE**: You can alternatively reject the order by specifying `"approved": false` in the request body.
Expected result (abbreviated):
```http
HTTP/1.1 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: 42
Approval sent for order: order_riley_o066i
```
1. Check the status of the workflow again to confirm that it completed successfully:
```sh
curl -i http://localhost:3000/orders/order_riley_o066i
```
Expected result (formatted for readability):
```json
{
"created_time": "2023-10-28T17:02:05.137170",
"details": {
"customer": "Riley",
"id": "order_riley_o066i",
"items": ["milk", "bread", "iPhone"],
"total": 1299.0
},
"id": "order_riley_o066i",
"last_updated_time": "2023-10-28T17:10:02.716316",
"order_result": {
"id": "order_riley_o066i",
"message": "Order processed successfully",
"success": true
},
"status": "COMPLETED"
}
```
1. Check the notifications web UI again to see the updated notifications from the workflow. They should look something like this:
```text
2023-10-28T10:02:05-07:00 | order_riley_o066i | Received order for Riley: ['milk', 'bread', 'iPhone']. Total = 1299.0
2023-10-28T10:02:05-07:00 | order_riley_o066i | Reserved inventory for order: ['milk', 'bread', 'iPhone']
2023-10-28T10:02:05-07:00 | order_riley_o066i | Waiting for approval since order >= 1000.0. Deadline = 2023-10-29 17:02:05.
2023-10-28T10:10:00-07:00 | order_riley_o066i | Order was approved by Chris.
2023-10-28T10:10:01-07:00 | order_riley_o066i | Payment was processed successfully
2023-10-28T10:10:02-07:00 | order_riley_o066i | Order submitted for shipping
```
1. Check the Zipkin UI at http://localhost:9411. Click "Run Query" to list the recently captured traces and then select the most recent, which should have a span of several seconds. The resulting UI should look something like this:
![Zipkin UI for success](./zipkin-workflow-success.png)
> **NOTE**: This screenshot shows a different order processing workflow instance ID than the one used in this example.
> **NOTE**: When the `orchestration||process_order_workflow` span is selected, you can click the "SHOW ALL ANNOTATIONS" button to see more information about the workflow, including the specific time that the order was approved.
1. To stop all the services, press `Ctrl+C` in the terminal where the services are running.
If you want to exercise the error handling path of the workflow, you can put the shipping service in "maintainance mode" and then try to submit another order.
1. Put the shipping service in "maintainance mode" by sending a POST request to the `/shipping/deactivate` endpoint:
```sh
curl -i -X POST http://localhost:3004/shipping/deactivate
```
1. Start a new order processing workflow. For simplicity, we'll start an order with a smaller cost that can be auto-approved.
```sh
curl -i -X POST http://localhost:3000/orders \
-H "Content-Type: application/json" \
-d '{"customer": "Casey", "items": ["milk", "bread"], "total": 10.00}'
```
1. Check the notifications web UI to see the notifications for this new workflow. It should look something like this:
```text
2023-10-28T10:46:19-07:00 | order_casey_1irq1 | Received order for Casey: ['milk', 'bread']. Total = 10.0
2023-10-28T10:46:20-07:00 | order_casey_1irq1 | Reserved inventory for order: ['milk', 'bread']
2023-10-28T10:46:21-07:00 | order_casey_1irq1 | Payment was processed successfully
2023-10-28T10:46:21-07:00 | order_casey_1irq1 | Error submitting order for shipping: order_casey_1irq1: Activity task #6 failed: Unknown Dapr Error. HTTP status code: 503
```
> **NOTE**: There is currently a bug in the v0.2.0 release of the Dapr Workflow Python SDK that causes the workflow to fail earlier than it's supposed to, resulting in the refund activity not running. This bug will be fixed in a future release of the SDK.
1. You can also check the status of the workflow using the REST API to see that the order is in the failed state.
```
curl -i http://localhost:3000/orders/order_casey_1irq1
```
Expected result (formatted for readability):
```json
{
"created_time": "2023-10-28T17:46:19.031582",
"details": {
"customer": "Casey",
"id": "order_casey_1irq1",
"items": ["milk", "bread"],
"total": 10.0
},
"failure_details": {
"error_type": "TaskFailedError",
"message": "order_casey_1irq1: Activity task #6 failed: Unknown Dapr Error. HTTP status code: 503",
"stack_trace": "..."
},
"id": "order_casey_1irq1",
"last_updated_time": "2023-10-28T17:46:21.205560",
"status": "FAILED"
}
```
1. You can also see the failed workflow in Zipkin, which should look something like this:
![Zipkin UI for failure](./zipkin-workflow-failure.png)

View File

@ -0,0 +1,12 @@
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: orders
spec:
type: pubsub.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""

View File

@ -0,0 +1,14 @@
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
- name: actorStateStore
value: "true"

View File

@ -0,0 +1,38 @@
version: 1
common:
resourcesPath: ./components
apps:
# The order processing service receives orders and executes workflows
- appDirPath: ./services/orderprocessing/
appPort: 3000
command: ["python3", "app.py"]
appLogDestination: console
daprdLogDestination: console
# The notifications service receives pubsub notifications from workflow
- appDirPath: ./services/notifications
appPort: 3001
command: ["python3", "app.py"]
appLogDestination: console
daprdLogDestination: console
# The inventory service is invoked directly by order processing workflows
- appDirPath: ./services/inventory
appPort: 3002
command: ["python3", "app.py"]
appLogDestination: console
daprdLogDestination: console
# The payments service is invoked directly by order processing workflows
- appDirPath: ./services/payments
appPort: 3003
command: ["python3", "app.py"]
appLogDestination: console
daprdLogDestination: console
# The shipping service is invoked directly by order processing workflows
- appDirPath: ./services/shipping
appPort: 3004
command: ["python3", "app.py"]
appLogDestination: console
daprdLogDestination: console

View File

@ -0,0 +1,30 @@
### Get the current inventory
GET http://localhost:3002/inventory
### Submit a simple order
POST http://localhost:3000/orders
Content-Type: application/json
{"customer": "Casey", "items": ["milk", "bread"], "total": 10.00}
### Submit an expensive order
POST http://localhost:3000/orders
Content-Type: application/json
{"customer": "Riley", "items": ["milk", "bread", "iPhone"], "total": 1299.00}
### Get the status of an order
GET http://localhost:3000/orders/order_casey_6glku
### Approve an order
POST http://localhost:3000/orders/order_riley_xcgyz/approve
Content-Type: application/json
{"approver": "Chris", "approved": true}

View File

@ -0,0 +1,90 @@
import logging
import os
import time
from typing import Dict
from flask import Flask, g, request
APP_PORT = os.getenv("APP_PORT", "3002")
app = Flask(__name__)
@app.route("/inventory", methods=["GET"])
def get_inventory():
logging.info("Getting inventory")
if not hasattr(g, 'inventory'):
restock_inventory()
inventory: Dict[str, int] = g.inventory
return inventory, 200
@app.route("/inventory/reserve", methods=["POST"])
def reserve_inventory():
logging.info(f"Reserving inventory: {request.json}")
order = request.json
items = order['items']
id = order['id']
if not hasattr(g, 'inventory'):
restock_inventory()
inventory: Dict[str, int] = g.inventory
# Check if we have enough inventory to fulfill the order
for item in items:
if item not in inventory or inventory[item] <= 0:
return {
"id": id,
"success": False,
"message": "Out of stock",
}, 200
# Update the inventory
for item in items:
inventory[item] = inventory[item] - 1
# Simulate work
time.sleep(1)
return {
"id": id,
"success": True,
"message": ""
}, 200
@app.route("/inventory/restock", methods=["POST"])
def restock_inventory():
logging.info("Restocking inventory")
inventory = dict[str, int](
{
'milk': 10,
'bread': 10,
'apples': 10,
'oranges': 10,
'iPhone': 10,
}
)
g.inventory = inventory
@app.route("/", methods=["GET"])
@app.route("/healthz", methods=["GET"])
def hello():
return f"Hello from {__name__}", 200
def main():
# Start the Flask app server
app.run(port=APP_PORT, debug=True, use_reloader=False)
app.post('/inventory/restock')
if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)03d %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO)
main()

View File

@ -0,0 +1,2 @@
dapr >= 1.11.0
flask >= 3.0.0

View File

@ -0,0 +1,64 @@
import logging
import os
from flask import Flask, jsonify, render_template, request
from flask_socketio import SocketIO
APP_PORT = os.getenv("APP_PORT", "3001")
PUBSUB_NAME = os.getenv("PUBSUB_NAME", "orders")
TOPIC_NAME = os.getenv("TOPIC_NAME", "notifications")
app = Flask(__name__)
socketio = SocketIO(app)
@socketio.on('connect')
def socket_connect():
print('connected', flush=True)
@app.route('/')
def index():
return render_template('index.html')
@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
"""Returns the list of topics the app wants to subscribe to.
Ref: https://docs.dapr.io/reference/api/pubsub_api/#provide-a-route-for-dapr-to-discover-topic-subscriptions"""
subs = [
{
'pubsubname': PUBSUB_NAME,
'topic': TOPIC_NAME,
'route': TOPIC_NAME,
},
]
return jsonify(subs)
@app.route('/' + TOPIC_NAME, methods=['POST', 'PUT'])
def topic_notifications():
"""Handles notification events from the Dapr pubsub component.
Ref: https://docs.dapr.io/reference/api/pubsub_api/#provide-routes-for-dapr-to-deliver-topic-events"""
logging.info(f"Received notification: {request.json}")
event = request.json
socketio.emit('message', event)
return '', 200
@app.route("/healthz", methods=["GET"])
def hello():
return f"Hello from {__name__}", 200
def main():
logging.info("Starting Flask app...")
socketio.run(app, port=APP_PORT, allow_unsafe_werkzeug=True)
if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)03d %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO)
main()

View File

@ -0,0 +1,2 @@
flask-socketio == 5.3.6
flask >= 3.0.0

View File

@ -0,0 +1,109 @@
/*
Knative colors
Dark Blue: #0865ad
Light Blue: #6695ca
*/
html,
body {
height: 100%;
margin: 0;
padding: 0;
background-color: #fff;
}
#wrapper {
padding: 0;
margin: 0;
height: 100%;
text-align: center;
widows: 100%;
}
#page-header {
padding: 0;
margin: 10px 0 0 0;
clear: both;
break-after: always;
height: 100px;
}
#page-header-image {
border: 0;
margin-left: 25px;
}
#page-header-image img {
width: 100px;
height: 100px;
float: left;
}
#connection {
float: right;
margin: 10px;
font-family: Geneva, Verdana, sans-serif;
font-size: 1em;
}
#middle-section {
margin: 0;
padding: 10px;
height: 700px;
overflow: auto;
}
#middle-section div {
margin: 5px;
}
#page-footer {
widows: 80%;
margin: 30px 0 0 0;
font-family: Geneva, Verdana, sans-serif;
font-size: 1em;
}
#notifications {
background-color: #20329b;
padding: 0;
margin: 5px;
height: 690px;
overflow: auto;
}
div.item-text {
margin-left: 10px;
padding-left: 10px;
}
#notifications b {
font-family: Geneva, Verdana, sans-serif;
font-size: 1em;
color: #20329b;
margin: 0;
}
#notifications i {
font-family: Geneva, Verdana, sans-serif;
font-size: 0.8em;
font-style: normal;
}
#notifications i.small {
font-size: 0.7em;
color: #666666;
}
.error {
color: red;
}
.item {
clear:both;
padding: 5px;
text-align: left;
background-color: #fff;
overflow: auto;
}

View File

@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<svg width="367px" height="270px" viewBox="0 0 367 270" version="1.1" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
<!-- Generator: Sketch 51.3 (57544) - http://www.bohemiancoding.com/sketch -->
<title>Artboard</title>
<desc>Created with Sketch.</desc>
<defs></defs>
<g id="Artboard" stroke="none" stroke-width="1" fill="none" fill-rule="evenodd">
<path d="M89.7917119,193.508761 L62.9000392,193.508761 L62.9000392,185.06412 C60.3311064,188.28895 57.7348835,190.639215 55.1112926,192.114985 C50.5200084,194.683918 45.3002339,195.968365 39.4518124,195.968365 C29.9959534,195.968365 21.5787254,192.716254 14.1998759,186.211935 C5.39991464,178.450478 1,168.174901 1,155.384895 C1,142.376257 5.50922929,131.991365 14.5278231,124.229909 C21.6880401,118.053538 29.9139674,114.9654 39.205852,114.9654 C44.6170083,114.9654 49.7001395,116.113203 54.4553981,118.408845 C57.1883053,119.720641 60.0031576,121.879605 62.9000392,124.885803 L62.9000392,70.3645766 L89.7917119,70.3645766 L89.7917119,193.508761 Z M63.7199073,155.466882 C63.7199073,150.656965 62.0255302,146.57133 58.6367253,143.209855 C55.2479203,139.848379 51.148621,138.167666 46.3387042,138.167666 C40.9822061,138.167666 36.5822914,140.189987 33.1388283,144.23469 C30.3512629,147.514178 28.9575012,151.258205 28.9575012,155.466882 C28.9575012,159.675559 30.3512629,163.419586 33.1388283,166.699075 C36.5276333,170.743777 40.9275479,172.766098 46.3387042,172.766098 C51.2032791,172.766098 55.3162428,171.09905 58.6777187,167.764903 C62.0391946,164.430756 63.7199073,160.331457 63.7199073,155.466882 Z M190.65006,193.508761 L163.758387,193.508761 L163.758387,185.06412 C161.189454,188.28895 158.593231,190.639215 155.969641,192.114985 C151.378356,194.683918 146.158582,195.968365 140.31016,195.968365 C130.854301,195.968365 122.437073,192.716254 115.058224,186.211935 C106.258263,178.450478 101.858348,168.174901 101.858348,155.384895 C101.858348,142.376257 106.367577,131.991365 115.386171,124.229909 C122.546388,118.053538 130.772315,114.9654 140.0642,114.9654 C145.475356,114.9654 150.558487,116.113203 155.313746,118.408845 C158.046653,119.720641 160.861506,121.879605 163.758387,124.885803 L163.758387,117.425004 L190.65006,117.425004 L190.65006,193.508761 Z M164.578255,155.466882 C164.578255,150.656965 162.883878,146.57133 159.495073,143.209855 C156.106268,139.848379 152.006969,138.167666 147.197052,138.167666 C141.840554,138.167666 137.440639,140.189987 133.997176,144.23469 C131.209611,147.514178 129.815849,151.258205 129.815849,155.466882 C129.815849,159.675559 131.209611,163.419586 133.997176,166.699075 C137.385981,170.743777 141.785896,172.766098 147.197052,172.766098 C152.061627,172.766098 156.174591,171.09905 159.536067,167.764903 C162.897543,164.430756 164.578255,160.331457 164.578255,155.466882 Z M294.54192,155.548869 C294.54192,168.557507 290.03269,178.942399 281.014097,186.703856 C273.85388,192.880226 265.627952,195.968365 256.336068,195.968365 C250.924911,195.968365 245.84178,194.820561 241.086522,192.524919 C238.353614,191.213123 235.538762,189.054159 232.64188,186.047961 L232.64188,231.550639 L205.750208,231.550639 L205.750208,117.425004 L232.64188,117.425004 L232.64188,125.869645 C235.046839,122.699473 237.643062,120.349208 240.430627,118.81878 C245.021911,116.249847 250.241686,114.9654 256.090107,114.9654 C265.545966,114.9654 273.963194,118.21751 281.342044,124.72183 C290.142005,132.483286 294.54192,142.758863 294.54192,155.548869 Z M266.584418,155.466882 C266.584418,151.148889 265.217985,147.404862 262.485078,144.23469 C259.041615,140.189987 254.614372,138.167666 249.203215,138.167666 C244.338641,138.167666 240.225677,139.834714 236.864201,143.168861 C233.502725,146.503008 231.822012,150.602307 231.822012,155.466882 C231.822012,160.276799 233.516389,164.362434 236.905194,167.72391 C240.293999,171.085386 244.393299,172.766098 249.203215,172.766098 C254.614372,172.766098 259.014286,170.743777 262.403091,166.699075 C265.190657,163.419586 266.584418,159.675559 266.584418,155.466882 Z M363.671373,142.267006 C359.899961,140.463288 356.073949,139.561442 352.19322,139.561442 C343.338601,139.561442 337.599582,143.168825 334.975991,150.3837 C333.992144,153.007291 333.500228,156.532689 333.500228,160.959998 L333.500228,193.508761 L306.608556,193.508761 L306.608556,117.425004 L333.500228,117.425004 L333.500228,129.886998 C336.342452,125.459689 339.403262,122.262235 342.682751,120.294542 C347.11006,117.670951 352.357164,116.359175 358.424218,116.359175 C359.845329,116.359175 361.594364,116.441161 363.671373,116.605136 L363.671373,142.267006 Z" id="dapr" fill="#0D2192"></path>
<polygon id="tie" fill="#0D2192" fill-rule="nonzero" points="205.538409 194.062172 232.614551 194.062172 234.946621 257.633831 219.07648 268.75443 203.206339 257.633831"></polygon>
<rect id="Rectangle-4" fill="#0D2192" fill-rule="nonzero" x="144.829497" y="2.27908829" width="102.722643" height="72.2941444" rx="2"></rect>
<rect id="Rectangle-4" fill="#FFFFFF" fill-rule="nonzero" opacity="0.0799999982" x="144.829497" y="2.27908829" width="37.9976369" height="72.2941444"></rect>
<rect id="Rectangle-3" fill="#0D2192" fill-rule="nonzero" x="112.390768" y="69.9090944" width="166.248488" height="17.3513412" rx="3.72016"></rect>
<rect id="Rectangle-4" fill="#FFFFFF" fill-rule="nonzero" opacity="0.0799999982" x="112.390768" y="69.9090944" width="51.4375478" height="21.3554969"></rect>
</g>
</svg>

After

Width:  |  Height:  |  Size: 5.4 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 115 KiB

View File

@ -0,0 +1,44 @@
window.onload = function () {
console.log("Protocol: " + location.protocol);
var wsURL = location.protocol + "//" + document.location.host
var log = document.getElementById("notifications");
function appendLog(item) {
var doScroll = log.scrollTop > log.scrollHeight - log.clientHeight - 1;
log.appendChild(item);
if (doScroll) {
log.scrollTop = log.scrollHeight - log.clientHeight;
}
}
if (log) {
var sock = io.connect(wsURL);
var connDiv = document.getElementById("connection-status");
connDiv.innerText = "closed";
sock.on('connect', function () {
console.log("connected to " + wsURL);
connDiv.innerText = "open";
});
sock.on('disconnect', function (e) {
console.log("connection closed (" + e.code + ")");
connDiv.innerText = "closed";
});
sock.on('message', function (evt) {
////console.log("message: " + JSON.stringify(evt));
var item = document.createElement("div");
item.className = "item";
var data = JSON.parse(evt.data);
var message = "<i>" + evt.time + "</i> | <b>" + data.order_id + "</b> | <i>" + data.message + "</i>";
var content = "<div class='item-text'>" + message + "</div>";
item.innerHTML = content;
appendLog(item);
});
} // if log
};

View File

@ -0,0 +1,43 @@
<!DOCTYPE html>
<html lang="en">
<head>
<title>dapr event viewer</title>
<meta charset="UTF-8">
<meta name="description" content="dapr event viewer">
<meta name="keywords" content="events, dapr, websocket, golang">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<link href="static/img/favicon.ico" rel="shortcut icon" />
<link rel="stylesheet" href="static/css/app.css" />
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.4.1/socket.io.js"></script>
<script type="text/javascript" src="static/js/app.js"></script>
</head>
<body>
<div id="wrapper">
<!-- Header -->
<div id="page-header">
<div id="page-header-image">
<img src="static/img/dapr.svg" alt="dapr logo" />
</div>
<div id="connection">connection: <b id="connection-status"></b></div>
</div>
<div id="middle-section">
<div id="notifications"></div>
</div>
<!-- Footer -->
<div id="page-footer">
<a href="https://github.com/dapr/samples">
Dapr Samples
</a>
</div>
</div>
</body>
</html>

View File

@ -0,0 +1,282 @@
import json
import logging
import os
import random
import string
from dataclasses import dataclass
from datetime import timedelta
import dapr.ext.workflow as wf
from dapr.clients import DaprClient
from flask import Flask, request, url_for
from markupsafe import escape
APP_PORT = os.getenv("APP_PORT", "3000")
PUBSUB_NAME = os.getenv("PUBSUB_NAME", "orders")
TOPIC_NAME = os.getenv("TOPIC_NAME", "notifications")
APPROVAL_THRESHOLD = 1000.0
APPROVAL_TIMEOUT = timedelta(hours=24)
app = Flask(__name__)
@dataclass
class Order:
id: str
customer: str
items: list
total: float
@dataclass
class OrderResult:
id: str
success: bool
message: str
@dataclass
class InventoryResult:
id: str
success: bool
message: str
@dataclass
class Approval:
approver: str
approved: bool
@staticmethod
def from_dict(dict):
return Approval(**dict)
def process_order_workflow(ctx: wf.DaprWorkflowContext, order: Order):
yield ctx.call_activity(notify, input=f"Received order for {order.customer}: {order.items}. Total = {order.total}")
# Call into the inventory service to reserve the items in this order
result = yield ctx.call_activity(reserve_inventory, input=order)
if not result.success:
yield ctx.call_activity(notify, input=f"Inventory failed for order: {result.message}")
return OrderResult(order.id, False, result.message)
yield ctx.call_activity(notify, input=f"Reserved inventory for order: {order.items}")
# Orders over $1,000 require human approval
if order.total >= APPROVAL_THRESHOLD:
approval_deadline = ctx.current_utc_datetime + APPROVAL_TIMEOUT
yield ctx.call_activity(
notify,
input=f"Waiting for approval since order >= {APPROVAL_THRESHOLD}. Deadline = {approval_deadline}.")
# Block the workflow on either an approval event or a timeout
approval_task = ctx.wait_for_external_event("approval")
timeout_expired_task = ctx.create_timer(approval_deadline)
winner = yield wf.when_any([approval_task, timeout_expired_task])
if winner == timeout_expired_task:
message = "Approval deadline expired."
yield ctx.call_activity(notify, input=message)
return OrderResult(order.id, False, message)
# Check the approval result
approval: Approval = yield approval_task
if not approval.approved:
message = f"Order was rejected by {approval.approver}."
yield ctx.call_activity(notify, input=message)
return OrderResult(order.id, False, message)
yield ctx.call_activity(notify, input=f"Order was approved by {approval.approver}.")
# Submit the order to the payment service
yield ctx.call_activity(submit_payment, input=order)
yield ctx.call_activity(notify, input="Payment was processed successfully")
# Submit the order for shipping
try:
yield ctx.call_activity(submit_order_to_shipping, input=order)
except Exception as e:
# Shipping failed, so we need to refund the payment
yield ctx.call_activity(notify, input=f"Error submitting order for shipping: {str(e)}")
yield ctx.call_activity(refund_payment, input=order)
yield ctx.call_activity(notify, input="Payment refunded")
# Allow the workflow to fail with the original failure details
raise
yield ctx.call_activity(notify, input="Order submitted for shipping")
return OrderResult(order.id, True, "Order processed successfully")
def notify(ctx: wf.WorkflowActivityContext, message: str):
logging.info(f"Sending notification: {message}")
with DaprClient() as d:
d.publish_event(PUBSUB_NAME, TOPIC_NAME, json.dumps({
"order_id": ctx.workflow_id,
"message": message
}))
def reserve_inventory(_, order: Order) -> InventoryResult:
logging.info(f"Reserving inventory for order: {order}")
with DaprClient() as d:
resp = d.invoke_method("inventory", "inventory/reserve", http_verb="POST", data=json.dumps(order.__dict__))
if resp.status_code != 200:
raise Exception(f"Error calling inventory service: {resp.status_code}")
inventory_result = InventoryResult(**json.loads(resp.data.decode("utf-8")))
logging.info(f"Inventory result: {inventory_result}")
return inventory_result
def submit_payment(_, order: Order):
logging.info(f"Submitting payment for order: {order}")
with DaprClient() as d:
resp = d.invoke_method("payments", "payments/charge", http_verb="POST", data=json.dumps(order.__dict__))
if resp.status_code != 200:
raise Exception(f"Error calling payment service: {resp.status_code}: {resp.text()}")
def submit_order_to_shipping(_, order: Order):
logging.info(f"Submitting order to shipping: {order}")
with DaprClient() as d:
resp = d.invoke_method("shipping", "shipping/ship", http_verb="POST", data=json.dumps(order.__dict__))
if resp.status_code != 200:
raise Exception(f"Error calling shipping service: {resp.status_code}: {resp.text()}")
def refund_payment(_, order: Order):
logging.info(f"Refunding payment for order: {order}")
with DaprClient() as d:
resp = d.invoke_method("payments", "payments/refund", http_verb="POST", data=json.dumps(order.__dict__))
if resp.status_code != 200:
raise Exception(f"Error calling payment service: {resp.status_code}: {resp.text()}")
# API to submit a new order
@app.route("/orders", methods=["POST"])
def submit_order():
request_data = request.get_json()
if not request_data:
return """Invalid request. Should be in the form of {
\"customer\": \"joe\", \"items\": [\"apples\", \"oranges\"], \"total\": 100.0}""", 400
if not request_data.get("customer"):
return "Missing customer name", 400
if not request_data.get("items"):
return "Missing items", 400
if not request_data.get("total"):
return "Missing total", 400
order = Order(
None,
request_data.get("customer"),
request_data.get("items"),
request_data.get("total"))
# Generate a unique ID for this order
random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=5))
order.id = f"order_{order.customer.lower()}_{random_suffix}"
wf_client = wf.DaprWorkflowClient()
instance_id = wf_client.schedule_new_workflow(
process_order_workflow,
input=order,
instance_id=order.id)
logging.info(f"Started workflow instance: {instance_id}")
return f"Order received. ID = '{escape(instance_id)}'", 202, {
'Location': url_for('check_order_status', order_id=instance_id, _external=True)
}
@app.route("/orders/<order_id>", methods=["GET"])
def check_order_status(order_id):
wf_client = wf.DaprWorkflowClient()
state = wf_client.get_workflow_state(order_id)
if not state:
return f"Order not found: {escape(order_id)}", 404
order_info = json.loads(state.serialized_input)
order = Order(
order_info.get('id'),
order_info.get('customer'),
order_info.get('items'),
order_info.get('total'))
resp = {
"id": state.instance_id,
"details": order.__dict__,
"status": state.runtime_status.name,
"created_time": state.created_at.isoformat(),
"last_updated_time": state.last_updated_at.isoformat(),
}
if state.serialized_output:
order_result_details = json.loads(state.serialized_output)
order_result = OrderResult(
order_result_details.get('id'),
order_result_details.get('success'),
order_result_details.get('message'))
resp["order_result"] = order_result.__dict__
if state.failure_details:
resp["failure_details"] = {
"message": state.failure_details.message,
"error_type": state.failure_details.error_type,
"stack_trace": state.failure_details.stack_trace
}
return resp, 200
@app.route("/orders/<order_id>/approve", methods=["POST"])
def approve_order(order_id):
request_data = request.get_json()
if not request_data:
return """Invalid request. Should be in the form of { \"approver\": \"joe\", \"approved\": true }""", 400
if not request_data.get("approver"):
return "Missing approver name", 400
if "approved" not in request_data:
return "Missing approved flag", 400
approval = Approval(
request_data.get("approver"),
request_data.get("approved"))
wf_client = wf.DaprWorkflowClient()
wf_client.raise_workflow_event(order_id, "approval", data=approval)
return f"Approval sent for order: {escape(order_id)}", 200
@app.route("/", methods=["GET"])
@app.route("/healthz", methods=["GET"])
def hello():
return f"Hello from {__name__}", 200
def main():
# Start the workflow runtime
logging.info("Starting workflow runtime...")
wf_runtime = wf.WorkflowRuntime() # host/port comes from env vars
wf_runtime.register_workflow(process_order_workflow)
wf_runtime.register_activity(notify)
wf_runtime.register_activity(reserve_inventory)
wf_runtime.register_activity(submit_payment)
wf_runtime.register_activity(submit_order_to_shipping)
wf_runtime.register_activity(refund_payment)
wf_runtime.start() # non-blocking
# Start the Flask app server
app.run(port=APP_PORT, debug=False, use_reloader=False)
# Stop the workflow runtime to allow the process to terminate
wf_runtime.shutdown()
if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)03d %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO)
main()

View File

@ -0,0 +1,2 @@
dapr-ext-workflow >= 0.2.0
flask >= 3.0.0

View File

@ -0,0 +1,48 @@
import logging
import os
import time
from flask import Flask, request
APP_PORT = os.getenv("APP_PORT", "3003")
app = Flask(__name__)
@app.route("/payments/charge", methods=["POST"])
def charge():
logging.info(f"Charging payment for order: {request.json}")
# Simulate work
time.sleep(1)
return '', 200
@app.route("/payments/refund", methods=["POST"])
def refund():
logging.info(f"Refunding payment for order: {request.json}")
# Simulate work
time.sleep(1)
return '', 200
@app.route("/", methods=["GET"])
@app.route("/healthz", methods=["GET"])
def hello():
return f"Hello from {__name__}", 200
def main():
# Start the Flask app server
app.run(port=APP_PORT, debug=True, use_reloader=False)
if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)03d %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO)
main()

View File

@ -0,0 +1,2 @@
dapr >= 1.11.0
flask >= 3.0.0

View File

@ -0,0 +1,58 @@
import logging
import os
import time
from flask import Flask, request
APP_PORT = os.getenv("APP_PORT", "3004")
app = Flask(__name__)
is_deactivated = False
@app.route("/shipping/ship", methods=["POST"])
def ship():
if is_deactivated:
return "The shipping service is currently deactivated for routine maintenance.", 503
logging.info(f"Shipping order: {request.json}")
# Simulate work
time.sleep(1)
return '', 200
@app.route("/shipping/deactivate", methods=["POST"])
def deactivate():
global is_deactivated
is_deactivated = True
logging.warning("The shipping service has been deactivated for routine maintenance.")
return '', 200
@app.route("/shipping/activate", methods=["POST"])
def activate():
global is_deactivated
is_deactivated = False
logging.info("The shipping service has been (re)activated.")
return '', 200
@app.route("/", methods=["GET"])
@app.route("/healthz", methods=["GET"])
def hello():
return f"Hello from {__name__}", 200
def main():
# Start the Flask app server
app.run(port=APP_PORT, debug=True, use_reloader=False)
if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)03d %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.INFO)
main()

View File

@ -0,0 +1,2 @@
dapr >= 1.11.0
flask >= 3.0.0

Binary file not shown.

After

Width:  |  Height:  |  Size: 126 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 123 KiB