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

Question

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

有没有人可以告诉我如何在prefect中创建审批工作流？

谢谢

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

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

使用信号

或者，如果我们使用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