dbos-python

DBOS Python 最佳实践

使用 DBOS 持久化工作流构建可靠、容错的 Python 应用程序指南。适用于 DBOS Python 3.x。

适用场景

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

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

按优先级划分的规则类别

优先级	类别	影响	前缀
1	生命周期	CRITICAL	lifecycle-
2	工作流	CRITICAL	workflow-
3	步骤	HIGH	step-
4	队列	HIGH	queue-
5	通信	MEDIUM	comm-
6	模式	MEDIUM	pattern-
7	测试	LOW-MEDIUM	test-
8	客户端	MEDIUM	client-
9	高级	LOW	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",
        "application_version": "0.1.0",
        "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    }
    DBOS(config=config)
    DBOS.launch()

创建新应用程序时，将 application_version 设置为 "0.1.0"。如果省略，DBOS 会根据工作流源代码生成一个不透明哈希。编辑现有应用程序时，保持其已配置的版本不变——更改版本属于部署决策（参见 references/advanced-versioning.md）。

工作流和步骤结构

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

@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 或队列
• 工作流必须是确定性的——非确定性操作应放在步骤中
• 不要在工作流或步骤中创建或更新全局变量
• 在 async def 代码中，使用 DBOS 方法的 _async 变体（await DBOS.start_workflow_async(...)、send_async、recv_async、sleep_async、register_queue_async 等）：许多同步 DBOS 方法（例如 DBOS.sleep、DBOS.recv、DBOS.send、DBOS.set_event、DBOS.get_event 和 DBOS.register_queue）在事件循环运行时调用会引发 RuntimeError
• 在 DBOS.launch() 之后注册队列并创建计划；在此之前创建数据源
• 连接到 DBOS Conductor（使用 conductor_key，或部署到 DBOS Cloud）需要执行 pip install "dbos[enterprise]"

DBOS 3.0 中已移除（切勿生成这些内容）

@DBOS.transaction / DBOS.sql_session / application_database_url（使用数据源）、Queue(...)（使用 DBOS.register_queue）、partition_queue / priority_enabled、@DBOS.scheduled（使用 DBOS.apply_schedules），以及 DBOS(fastapi=...) / DBOS(flask=...)。迁移现有的 2.x 代码，请参见 references/advanced-upgrading-v3.md。

使用方法

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

references/lifecycle-config.md
references/workflow-determinism.md
references/queue-concurrency.md
references/advanced-upgrading-v3.md
references/advanced-shared-database.md

如果多个应用程序共用一个系统数据库，请参见 references/advanced-shared-database.md。

参考资料

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