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

问题描述

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

谁能告诉我如何才能完美地创建审批工作流程？

谢谢

解决方法

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

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

更多资源：

prefect