Joël Luijmes
02/04/2021, 10:41 AMimport prefect
from prefect import task, Flow, case
from prefect.tasks.control_flow import merge
@task
def check_condition():
return True
@task
def log_output(val):
<http://prefect.context.logger.info|prefect.context.logger.info>(val)
with Flow("conditional-branches") as flow:
cond = check_condition()
with case(cond, False):
task_true = log_output("Branch: true")
task_end = log_output("Main branch end")
task_end.set_upstream(merge(task_true))
flow.visualize()
flow.run()
task_end.skip_on_upstream_skip = False
works, but is it the correct way?josh
02/04/2021, 5:08 PMJoël Luijmes
02/05/2021, 7:13 AMtask_end = log_output("Main branch end")
task_end.skip_on_upstream_skip = False
task_end.set_upstream(task_true)
josh
02/05/2021, 2:53 PMskip_on_upstream_skip
is definitely a suitable approach for this use case!