使用 DBOS 构建高可靠且容错的 Python 应用

通过引入 DBOS 持久化工作流，解决 Python 应用在处理复杂业务逻辑时容易出现的状态丢失和崩溃问题，实现自动恢复和强一致性的分布式执行。

为什么需要这个技能

在开发复杂的分布式系统或长耗时任务时，传统的 Python 应用一旦崩溃，内存状态会全部丢失，且很难在中断点精确恢复。

DBOS 提供了“持久化工作流”的概念，将工作流拆分为一系列可重试的“步骤（Steps）”。它会自动记录每个步骤的执行结果，如果程序在执行过程中崩溃，DBOS 能够确保在重启后跳过已完成的步骤，直接从失败点继续执行，从而构建出具备天然容错能力的系统。

适用场景

• 现有 Python 项目升级：需要为现有业务代码增加状态持久化和自动恢复能力。
• 复杂异步工作流：涉及多个外部 API 调用，且要求在调用失败时能够可靠重试。
• 并发控制需求：需要通过队列（Queues）严格控制并发执行的频率或顺序。
• 分布式通信：需要在不同工作流之间实现基于事件、消息或流的通信。

核心工作流

1. 配置与启动

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()

2. 拆分工作流与步骤

核心原则是：工作流（Workflow）负责编排逻辑，步骤（Step）负责执行副作用。所有访问外部服务或执行复杂操作的代码必须定义为 step。

@DBOS.step()
def call_external_api():
    # 外部 API 调用放在 step 中，以便 DBOS 记录结果
    return requests.get("https://api.example.com").json()

@DBOS.workflow()
def my_workflow():
    # 工作流调用步骤
    result = call_external_api()
    return result

3. 关键约束

为了保证容错机制生效，必须遵循以下开发禁忌：

• 保证确定性：工作流必须是确定性的，所有非确定性操作（如获取当前时间、随机数、网络请求）必须放在 step 中。
• 严禁在 Step 中启动工作流：不要在 step 内部调用 DBOS.start_workflow 或 DBOS.recv。
• 避免全局变量：禁止在工作流或步骤中创建/更新全局变量。
• 禁止多线程启动：不要使用 Python 原生线程启动工作流，应使用 DBOS.start_workflow 或队列。

下载和安装

下载 dbos-python 中文版 Skill ZIP

解压后将目录放入你的 AI 工具 skills 文件夹，重启工具后即可使用。具体路径参考内附的 USAGE.zh.md。

你可能还需要

暂无推荐