dbos-inc/dbosify-py

来源：https://github.com/dbos-inc/dbosify-py

DBOSify

DBOSify 是 Temporal Python (https://github.com/temporalio/sdk-python) 的即插即用替代方案，它使用 Postgres（通过 DBOS Transact (https://github.com/dbos-inc/dbos-transact-py)）来代替 Temporal 服务器。这样就可以在无需任何基础设施（Postgres 除外）的情况下运行持久化工作流、活动、信号、更新、重试和恢复。

使用此库时，请将 dbosify 导入以替代 temporalio，并将工作进程和客户端连接到 Postgres 数据库：

使用方式

安装：

pip install dbosify

这是一个即插即用的替代方案：只需导入 dbosify 而非 temporalio，并将客户端和工作进程连接到 Postgres 数据库而非 Temporal 服务器。更多文档请参考这里 (https://docs.dbos.dev/explanations/dbosify)。

import asyncio
import os
from datetime import timedelta

from dbosify import activity, workflow
from dbosify.client import Client
from dbosify.worker import Worker

# 设置为你 Postgres 数据库的连接字符串
DB_URL = os.environ.get("DBOS_SYSTEM_DATABASE_URL")

@activity.defn
async def compose_greeting(name: str) -> str:
    return f"Hello, {name}!"

@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            compose_greeting,
            name,
            start_to_close_timeout=timedelta(seconds=10)
        )

async def main() -> None:
    worker = Worker(
        DB_URL,
        task_queue="greetings",
        workflows=[GreetingWorkflow],
        activities=[compose_greeting],
    )
    async with worker:
        async with await Client.connect(DB_URL) as client:
            result = await client.execute_workflow(
                GreetingWorkflow.run,
                "World",
                id="greeting-1",
                task_queue="greetings"
            )
    print(result)  # Hello, World!

if __name__ == "__main__":
    asyncio.run(main())

工作原理

DBOSify 将每个 Temporal 工作流作为基于 Postgres 的 DBOS 工作流运行。一个确定性解释器在虚拟事件循环上运行该工作流（包括其主协程、信号、更新和查询处理程序），该事件循环仅在有事件到达时才会推进。利用 DBOS 步骤和工作流通信原语 (https://docs.dbos.dev/python/tutorials/workflow-communication)，所有非确定性操作都会在工作流观察到之前被检查点到 Postgres 中。

• 活动和定时器 成为 DBOS 步骤和持久化休眠，每一步在完成时都会被检查点保存。
• 信号、更新和取消 是通过 Postgres 的 LISTEN/NOTIFY 传递的持久化消息。
• 恢复 在新的工作进程上重新运行工作流：解释器针对已记录的检查点重放相同序列的操作，因此执行将从上次中断处恢复并恰好完成一次。
• 命名空间 各自映射到自己的 Postgres 模式；Client 封装了 DBOS 客户端，Worker 封装了 DBOS 运行时。

测试方式

由于 DBOSify 是 Temporal 的即插即用替代方案，其测试涵盖了正确性和与 Temporal 的一致性，采用以下策略：

• 移植所有相关的 Temporal Python 单元测试和集成测试
• 移植相关的 Temporal Python 示例应用，验证 DBOSify 是即插即用的替代方案
• 新增单元测试和集成测试，重点是通过 kill-and-recover 测试验证确定性故障恢复
• 签名一致性测试，机械地断言这些库的公共 API 是相同的（有文档记录的特例除外）

这不是什么

• 没有有线协议兼容性。 没有 gRPC 有线兼容性。其他语言的 Temporal SDK 无法连接。此方案仅针对纯 Python 应用，完全替换了 Temporal 服务器和 Python SDK。
• 没有 Temporal Web UI、temporal CLI 或 tctl。 你可以使用 DBOS 的工作流管理 API 和 DBOS Conductor 来操作工作流。有关架构差异和功能兼容性的信息，请参见 此文档。