Hi can someone help me understand what the arg on triggers are used for i.e. upstream_states. For example,

*prefect.triggers.all_finished*(_upstream_states_)

. This is for prefect 1 btw.

hi @Michael Z - it might be helpful to look at an example

from prefect import Flow, task, triggers

@task
def something_that_succeeds():
    print("I succeed!")
@task
def something_that_fails():
    raise ValueError("I fail!") 

@task(trigger=triggers.all_finished)
def as_long_as_we_get_here():
    print(".. I'll run!")

@task(trigger=triggers.all_successful)
def only_if_all_upstream_succeed():
    print(".. only then, will I run")

with Flow('My Flow') as flow:
    a = something_that_succeeds() 
    b = something_that_fails()
    c = as_long_as_we_get_here(upstream_tasks=[a, b]) 
    d = only_if_all_upstream_succeed(upstream_tasks=[a, b])

if __name__ == "__main__":
   flow.run()

so pretty much

all_finished

says "run this task when all its upstream tasks have a terminal state, whether those states are Failed or Successful" in contrast to

all_successful

, which requires all upstream task states be in a successful state for that task to run + more of them