https://prefect.io logo
Title
i

itay livni

02/24/2020, 7:07 PM
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])
a

Alex Goodman

02/24/2020, 7:25 PM
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())
i

itay livni

02/24/2020, 7:28 PM
@Alex Goodman Thanks!. Let me try to reconcile what I have and your example
👍 1
a

Alex Goodman

02/24/2020, 7:31 PM
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
i

itay livni

02/24/2020, 7:40 PM
@Alex Goodman Yay! The mistake was I was passing a list to the
List
a

Alex Goodman

02/24/2020, 7:40 PM
ahh! good catch!