如何使用 prefect 创建审批工作流？

Question

我已经开始在 python 中完善工作流程，但没有找到有关审批工作流程的任何帮助。

谁能让我知道我将如何以完美的方式创建审批工作流？

谢谢

Answer 1

在 Prefect 中有两种配置审批工作流的方法。

使用触发器

触发器允许您指定任务运行所需的基于状态的条件；在审批的情况下，我们可以使用manual_only触发器，这样每次这个任务运行并且它的上游完成时，任务就会进入Paused状态并等待审批：

from prefect import task, Flow
from prefect.triggers import manual_only

@task
def build():
    print("build task")
    return True

@task
def test(build_result):
    print("test task")
    build_result == True

@task(trigger=manual_only)
def deploy():
    """
    With the manual_only trigger this task will only run after it has been approved
    """
    print("deploy task")
    pass
    
with Flow("code_deploy") as flow:
    res = build()
    deploy(upstream_tasks=[test(res)])

请注意，使用触发器意味着每次此流程运行时，部署任务都会等待批准。

使用信号

另外，如果我们使用Prefect Signal，我们可以获得一些灵活性；信号允许我们通过特殊类型的异常强制任务运行进入某些状态。在这种情况下，我们只能在满足某些条件时暂停等待：

from prefect import task, Flow
from prefect.engine.signals import PAUSE

@task
def build():
    print("build task")
    return True

@task
def test(build_result):
    print("test task")
    build_result == True

@task
def deploy():
    """
    With the manual_only trigger this task will only run after it has been approved
    """
    print("deploy task")
    if some_condition:
        raise PAUSE("Condition met - waiting for approval.")
    pass

with Flow("code_deploy") as flow:
    res = build()
    deploy(upstream_tasks=[test(res)])

这允许更细粒度地控制暂停发生的时间，甚至暂停发生的时间（您可以在初始化PAUSE 信号时指定start_time，这样就无需手动干预）。

更多资源：

• Blog Post
• Idiom in the documentation