JC Garcia
02/19/2021, 3:28 PMany_failed
trigger is not fired because no task failed. Is there any way to fix this? docsnicholas
JC Garcia
02/19/2021, 4:01 PM@task
def task_one:
#no failure here
@task
def task_two:
#no failure here
@task(trigger=any_failed)
def set_failure() -> None:
raise signals.FAIL()
with Flow("flow-name") as flow:
task_one()
task_two()
failure = set_failure(upstream_tasks=[task_one, task_one])
nicholas
set_reference_tasks
to a list of the tasks that you want to determine the final state of your flow. So you could so something like this:
@task
def task_one:
#no failure here
@task
def task_two:
#no failure here
@task(trigger=any_failed)
def set_failure() -> None:
raise signals.FAIL()
with Flow("flow-name") as flow:
task_1 = task_one()
task_2 = task_two()
failure = set_failure(upstream_tasks=[task_one, task_one])
flow.set_reference_tasks([task_1, task_2])
JC Garcia
02/19/2021, 4:10 PM