Dapr Python SDK 与 Dapr 工作流扩展集成
如何快速上手使用 Dapr 工作流扩展
Dapr Python SDK 提供了一个内置的 Dapr 工作流扩展 dapr.ext.workflow，用于创建 Dapr 服务。
安装
您可以通过以下命令下载并安装 Dapr 工作流扩展：
pip install dapr-ext-workflow
Note
开发版包将包含与 Dapr 运行时预发布版本兼容的功能和行为。在安装 <code>dapr-dev</code> 包之前，请确保已卸载任何稳定版本的 Python SDK 扩展。
pip install dapr-ext-workflow-dev
示例
from time import sleep
import dapr.ext.workflow as wf
wfr = wf.WorkflowRuntime()
@wfr.workflow(name='random_workflow')
def task_chain_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
try:
result1 = yield ctx.call_activity(step1, input=wf_input)
result2 = yield ctx.call_activity(step2, input=result1)
except Exception as e:
yield ctx.call_activity(error_handler, input=str(e))
raise
return [result1, result2]
@wfr.activity(name='step1')
def step1(ctx, activity_input):
print(f'Step 1: Received input: {activity_input}.')
# Do some work
return activity_input + 1
@wfr.activity
def step2(ctx, activity_input):
print(f'Step 2: Received input: {activity_input}.')
# Do some work
return activity_input * 2
@wfr.activity
def error_handler(ctx, error):
print(f'Executing error handler: {error}.')
# Do some compensating work
if __name__ == '__main__':
wfr.start()
sleep(10) # wait for workflow runtime to start
wf_client = wf.DaprWorkflowClient()
instance_id = wf_client.schedule_new_workflow(workflow=task_chain_workflow, input=42)
print(f'Workflow started. Instance ID: {instance_id}')
state = wf_client.wait_for_workflow_completion(instance_id)
print(f'Workflow completed! Status: {state.runtime_status}')
wfr.shutdown()
后续步骤
Dapr 工作流 Python SDK 入门1 - Dapr Workflow Python SDK 入门
如何使用 Dapr Python SDK 快速上手工作流
让我们创建一个 Dapr 工作流并通过控制台调用它。借助提供的工作流示例，你将：
- 运行一个 Python 控制台应用程序，该程序演示包含活动、子工作流和外部事件的工作流编排
- 了解如何处理重试、超时以及工作流状态管理
- 使用 Python 工作流 SDK 来启动、暂停、恢复和清理工作流实例
本示例使用自托管模式下通过 dapr init 初始化的默认配置。
在 Python 示例项目中，simple.py 文件包含应用程序的设置，包括：
前置条件
设置环境
首先克隆 [Python SDK 仓库]。
git clone https://github.com/dapr/python-sdk.git
从 Python SDK 根目录导航到 Dapr Workflow 示例。
运行以下命令，安装使用 Dapr Python SDK 运行此工作流示例所需的所有依赖。
pip3 install -r workflow/requirements.txt
在本地运行应用程序
要运行 Dapr 应用程序，你需要启动 Python 程序和一个 Dapr 边车。在终端中运行：
dapr run --app-id wf-simple-example --dapr-grpc-port 50001 --resources-path components -- python3 simple.py
注意： 由于 Windows 上未定义 Python3.exe，你可能需要使用 python simple.py 而不是 python3 simple.py。
预期输出
- "== APP == Hi Counter!"
- "== APP == New counter value is: 1!"
- "== APP == New counter value is: 11!"
- "== APP == Retry count value is: 0!"
- "== APP == Retry count value is: 1! This print statement verifies retry"
- "== APP == Appending 1 to child_orchestrator_string!"
- "== APP == Appending a to child_orchestrator_string!"
- "== APP == Appending a to child_orchestrator_string!"
- "== APP == Appending 2 to child_orchestrator_string!"
- "== APP == Appending b to child_orchestrator_string!"
- "== APP == Appending b to child_orchestrator_string!"
- "== APP == Appending 3 to child_orchestrator_string!"
- "== APP == Appending c to child_orchestrator_string!"
- "== APP == Appending c to child_orchestrator_string!"
- "== APP == Get response from hello_world_wf after pause call: Suspended"
- "== APP == Get response from hello_world_wf after resume call: Running"
- "== APP == New counter value is: 111!"
- "== APP == New counter value is: 1111!"
- "== APP == Workflow completed! Result: "Completed"
发生了什么？
当你运行应用程序时，会演示几个关键的工作流功能：
工作流和活动注册：应用程序使用 Python 装饰器自动向运行时注册工作流和活动。这种基于装饰器的方法提供了一种简洁、声明式的方式来定义你的工作流组件：
@wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
# Workflow definition...
@wfr.activity(name='hello_act')
def hello_act(ctx: WorkflowActivityContext, wf_input):
# Activity definition...
运行时设置：应用程序初始化工作流运行时和客户端：
wfr = WorkflowRuntime()
wfr.start()
wf_client = DaprWorkflowClient()
活动执行：工作流执行一系列活动来递增计数器：
@wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
yield ctx.call_activity(hello_act, input=1)
yield ctx.call_activity(hello_act, input=10)
重试逻辑：工作流演示了使用重试策略进行错误处理：
retry_policy = RetryPolicy(
first_retry_interval=timedelta(seconds=1),
max_number_of_attempts=3,
backoff_coefficient=2,
max_retry_interval=timedelta(seconds=10),
retry_timeout=timedelta(seconds=100),
)
yield ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
子工作流：子工作流使用自己的重试策略执行：
yield ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)
外部事件处理：工作流等待一个带有超时的外部事件：
event = ctx.wait_for_external_event(event_name)
timeout = ctx.create_timer(timedelta(seconds=30))
winner = yield when_any([event, timeout])
工作流生命周期管理：示例演示如何暂停和恢复工作流：
wf_client.pause_workflow(instance_id=instance_id)
metadata = wf_client.get_workflow_state(instance_id=instance_id)
# ... check status ...
wf_client.resume_workflow(instance_id=instance_id)
事件触发：恢复后，工作流触发一个事件：
wf_client.raise_workflow_event(
instance_id=instance_id,
event_name=event_name,
data=event_data
)
完成与清理：最后，工作流等待完成并进行清理：
state = wf_client.wait_for_workflow_completion(
instance_id,
timeout_in_seconds=30
)
wf_client.purge_workflow(instance_id=instance_id)
后续步骤