Hi - I have

task

in a

Flow

that takes a

list

. How do I set rhe

List

Trigger

to run on

any_successful

upstream tasks? https://docs.prefect.io/core/concepts/tasks.html#collections I tried a couple approaches including def_list = List(trigger=any_successful) and then calling

munged_defs_df = definitions.munge_dfs(def_list([df1, df2, df3])

Interesting, as you described, it should be working. Here is a small repro attempt that appears to work as expected:

import prefect
from prefect.tasks.core.collections import List


@prefect.task()
def a_task():
    raise prefect.engine.signals.FAIL


@prefect.task()
def b_task():
    pass


@prefect.task()
def c_task():
    pass


@prefect.task()
def d_task():
    pass

def_list = List(trigger=prefect.triggers.any_successful)

with prefect.Flow("My Flow") as flow:
    flow.chain(a_task, def_list(b_task, c_task), d_task)

flow.visualize(flow.run())

@Alex Goodman Thanks!. Let me try to reconcile what I have and your example

righto, shout out if you find out what it is! Next steps I would take would be to take a closer look at the upstream states from

List

@Alex Goodman Yay! The mistake was I was passing a list to the

List

ahh! good catch!