0

我已经开始研究 python 中的完美工作流程，但没有找到任何有关批准工作流程的帮助。

谁能让我知道我将如何能够完美地创建审批工作流程？

谢谢

4

1 回答 1

0

在 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信号时指定 a，这样就不需要手动干预）。

更多资源：

于 2021-02-16T17:35:50.400 回答