背景介绍

如果工作中我们要运行的程序分几个步骤去执行，如果当中间某个步骤执行失败的时候，我们不得不重新全部执行一遍，其实这样很费事费力，因为每次都是从头开始执行，而且这种代码放到生产中也会出现不稳定的情况，毕竟基础硬件的底层也会出现问题，比如当网络、操作系统等出现问题时，你所定时执行的脚本或程序就会出现问题，针对此问题，我们今天就介绍一款轻量级能持久执行的Python库DBOS Transact 来解决这个问题。

持久执行意味着我们程序能够灵活应对任何故障，如果它被中断或崩溃，所有工作流将自动从最后完成的步骤恢复，无论我们尝试崩溃多少次，它总是从上次中断的地方恢复。

说一下这个库的简单原理，在后台DBOS Transact 的工作原理是将程序的执行状态，就是当前正在执行的工作流以及目前完成的步骤存储在了数据库中，因此我们只需要使用一个Postgres数据库即可，大家不要觉得安装数据库很复杂，现在已经非常简单了，使用docker分分钟搞定。

DBOS Transact还有一些更酷的功能，包括：

• 计划作业 ： 每个时间间隔恰好运行一次工作流。

• 恰好一次事件处理 ： 使用工作流处理传入事件（例如，来自 Kafka 主题）恰好一次。

• 可观测性 ： 所有工作流都会自动发出 OpenTelemetry 跟踪。

说了这么多，让我们动手吧
Let's 开始

先安装

pip install dbos
dbos init --config

然后，使用这个简单的程序（需要 Postgres）尝试一下：

from fastapi import FastAPI
from dbos import DBOS

app = FastAPI()
DBOS(fastapi=app)

@DBOS.step()
def step_one():
    print("Step one completed!")

@DBOS.step()
def step_two():
    print("Step two completed!")

@DBOS.workflow()
def dbos_workflow():
    step_one()
    for _ in range(5):
        print("Press Control + \ to stop the app...")
        DBOS.sleep(1)
    step_two()

@app.get("/")
def fastapi_endpoint():
    dbos_workflow()

将以上程序保存到文件中，配置连接到Postgres数据库，然后启动它。然后再浏览器中启动工作流，当出现提示时我们强制退出应用程序，结果如我们所料，它应该中途崩溃了，完成了第一步，但没有完成第二步，然后当我们重新启动程序，它会从中断的地方恢复直接完成第二步，以上就是基本的使用，更多的使用请参考文档：

https://docs.dbos.dev

​

原创 python运维技术