python - 如何进入知府核心本地任务最终状态？

Question

我已经构建了一个流程，如果 kwarg 为空，它会隐式跳过运行给定的任务。
我在任务函数中使用类似的东西来跳过逻辑:

if kwargs.get('processors', Hierarchy()).__len__() == 0:
                    raise signals.SKIP('skipping task',
                                       result=Prediction())    

我想构建一些单元测试以确保跳过所述任务的最终状态。在任务级别获取状态的最简单方法是什么？
我可以从文档中看到如何获取流程而不是任务。
更新
为了补充克里斯的回应，我使用了他提出的第一个选项。由于我的流程是在测试之外定义的，因此我创建了一个简单的函数来获取一组跳过的任务。在测试中，这与应该跳过的任务列表进行了比较:

def get_skipped_tasks(flow_state):
        return set(key.name for key, value in flow_state.result.items() if value.is_skipped())

Answer 1

为了完整起见，我将在这里包括几种方法；对于我的示例，我将使用这个基本流程:

from prefect import task, Flow
from prefect.engine.signals import SKIP
import random


@task
def random_number():
    return random.randint(0, 100)

@task
def is_even(num):
    if num % 2:
        raise SKIP("odd number")
    return True

with Flow("dummy") as flow:
    even_task = is_even(random_number)

运行整个流程
以交互方式运行时，您始终可以运行整个流程并从父流程运行状态访问各个任务状态；请注意，当您“调用”一个任务(例如， is_even(random_number) )时，会创建一个副本，因此您需要正确跟踪这些副本。

flow_state = flow.run()

assert flow_state.result[even_task].is_skipped() # for example

使用模拟数据运行一段流程
当以交互方式运行时，您还可以传递任务字典 -> 运行者将尊重的状态；可以选择向这些状态提供数据:

from prefect.engine.state import Success

mocked_state = Success(result=2)


flow_state = flow.run(task_states={random_number: mocked_state})
assert not flow_state.result[even_task].is_skipped()

使用 TaskRunner
最后，如果你想单独对这个任务运行基于状态的测试，你可以使用 TaskRunner .这变得有点复杂，因为您必须使用 Edge 重新创建上游依赖项s。

from prefect.engine.task_runner import TaskRunner
from prefect.edge import Edge

runner = TaskRunner(task=even_task)
edge = Edge(key="num", upstream_task=random_number, downstream_task=even_task)


task_state = runner.run(upstream_states={edge: mocked_state})
assert not task_state.is_skipped()