Michael Z11/18/2022, 10:30 PM
. This is for prefect 1 btw.
Nate11/19/2022, 5:27 AM
so pretty much
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()
says "run this task when all its upstream tasks have a terminal state, whether those states are Failed or Successful" in contrast to
, which requires all upstream task states be in a successful state for that task to run + more of them