Hello there! I have a task that most of the time I want to depend on a single upstream task. However, at the end of the flow, I want it to run if

any_successful

for several upstream tasks. When I write it like below,

task_5

doesn’t fail, even if

task_3

and

task_4

fail. Is it because the creation of the list

[task_3, task_4]

succeeds? How should I write this flow instead? Many thanks!

task_3 = my_task(
    "input1", task_args=dict(name="input1")
    ).set_upstream(task_1)

task_4 = my_task(
    "input2", task_args=dict(name="input2")
    ).set_upstream(task_2)

task_5 = my_task("all_inputs",
    task_args=dict(name="all_inputs", trigger=any_successful),
    ).set_upstream([task_3, task_4])

Hi Leanna, you are right, prefect generates a List task, and the

List

task probably runs even if all of its upstreams fail. You can validate that a List task is generated by running

flow.visualize()

task.set_upstream

is designed to handle a single upstream task, either call it multiple times, or use

task.set_dependencies()

instead.

# Multiple set_upstreams
task_5 = my_task("all_inputs",
    task_args=dict(name="all_inputs", trigger=any_successful)).set_upstream(task_3).set_upstream(task_4)

# set_dependencies
task_5 = my_task("all_inputs",
    task_args=dict(name="all_inputs", trigger=any_successful)).set_dependencies(upstream_tasks=[task_3, task_4])

thank you @emre!