In a flow I want to check a result, and only execu...
# prefect-community
j
In a flow I want to check a result, and only execute some downstream tasks if that condition is true. However, later on I want to merge that branch back in the main flow. E.g. https://docs.prefect.io/core/idioms/conditional.html#merging-branches-in-a-flow The example above deals with two branches, True/False, where I’m only interested in checking for True. I got the following POC, which works fine AFAIK, but I’d like to validate this is the rightt method, and if I’m not introducing undesired behavior. Edit: I got a better example which doesn’t work as I want to below. I always want to return to the main branch and thus run task_end. But this doesn’t work as I want, now everything is Skipped because the condition doesn’t work. Basically I want a task which continues on Skipped or Success I suppose? What is best practice here
Copy code
import 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()
Setting
task_end.skip_on_upstream_skip = False
works, but is it the correct way?
j
Hi @Joël Luijmes I think I’m following here. So after seeing your edit to the original message does the skip_on_upstream_skip do what you want?
j
Yes, it does what I want. Thus printing Main branch end, no matter if the task_true was skipped or not. But I’m not sure if this is the right approach?
Copy code
task_end = log_output("Main branch end")
    task_end.skip_on_upstream_skip = False
    task_end.set_upstream(task_true)
j
Yeah
skip_on_upstream_skip
is definitely a suitable approach for this use case!