dbos-python

DBOS Python 最佳实践

使用 DBOS 持久化工作流构建可靠、容错的 Python 应用指南。

适用场景

在以下情况下参考这些指南：

• 将 DBOS 添加到现有 Python 代码中
• 创建工作流和步骤
• 使用队列控制并发
• 实现工作流通信（事件、消息、流）
• 配置并启动 DBOS 应用
• 在外部应用中使用 DBOSClient
• 测试 DBOS 应用

运行时边界

示例是对版本敏感的代码片段，省略了应用设置、导入和处理程序；请确认已安装的 DBOS API。持久化工作流的标识并不能确保任意外部副作用恰好执行一次：在检查点写入之前发生崩溃后，步骤可能会重复执行。对于付款和消息，请使用提供商的幂等键并进行对账。授权数据库目标和写入操作，保护序列化输入和结果，切勿反序列化不可信的 pickle 数据。API 路由需要身份验证、租户所有权校验、Webhook 签名和重放验证，以及有界等待时间。执行会造成破坏的测试重置时，必须使用专用且可丢弃的系统数据库；仅凭变量名称无法证明目标正确。

按优先级划分的规则类别

优先级	类别	影响	前缀
1	生命周期	严重	lifecycle-
2	工作流	严重	workflow-
3	步骤	高	step-
4	队列	高	queue-
5	通信	中	comm-
6	模式	中	pattern-
7	测试	低至中	test-
8	客户端	中	client-
9	高级	低	advanced-

关键规则

DBOS 配置和启动

DBOS 应用必须在主函数中配置并启动 DBOS：

import os
from dbos import DBOS, DBOSConfig

@DBOS.workflow()
def my_workflow():
    pass

if __name__ == "__main__":
    config: DBOSConfig = {
        "name": "my-app",
        "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    }
    DBOS(config=config)
    DBOS.launch()

工作流和步骤结构

工作流由步骤组成。任何执行复杂操作或访问外部服务的函数都必须是一个步骤：

@DBOS.step()
def call_external_api():
    return requests.get("https://api.example.com").json()

@DBOS.workflow()
def my_workflow():
    result = call_external_api()
    return result

关键限制

• 不要在步骤中调用 DBOS.start_workflow 或 DBOS.recv
• 不要使用线程启动工作流——请使用 DBOS.start_workflow 或队列
• 工作流必须具有确定性——非确定性操作应放在步骤中
• 不要在工作流或步骤中创建或更新全局变量

使用方法

阅读各规则文件，了解详细说明和示例：

references/lifecycle-config.md
references/workflow-determinism.md
references/queue-concurrency.md

参考资料

• https://docs.dbos.dev/
• https://github.com/dbos-inc/dbos-transact-py

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为针对特定环境进行验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并请求澄清。