I’m trying to use a parameter as a condition in a flow. What I’m finding is that if the condition is False, then every task after it is skipped.

with Flow('MyFlow') as flow:
    do_thing = Parameter('do_think', default=False, required=False)

    data = get_data()

    with case(do_thing, True):
        data = do_thing_task(data)

    data = do_another_thing(data) # This gets skipped when do_thing = False

Hi Carl! You’re overloading the variable name

data

so you are probably creating a dependency graph that you don’t intend to create; you should use new variable names with each task call:

with Flow('MyFlow') as flow:
    do_thing = Parameter('do_think', default=False, required=False)

    data = get_data()

    with case(do_thing, True):
        true_data = do_thing_task(data)

    other_data = do_another_thing(data)

In that case, how do I pass

true_data

to

do_another_thing()

when the condition is True?

other_data = do_another_thing(true_data)

would do the trick, but in that case

do_another-thing

will skip if the condition is

False

, which is expected but sounds like not what you want?

Yeah that isn’t what I’m after. I could probably add a

with case(do_thing, False):
    true_data = data

but that feels pretty messy

when a case statement isn’t met, the tasks within it are skipped (that’s what case statements are for); the default behavior of Prefect is that if an upstream dependency skips, downstream dependencies will also skip. You can override this behavior with the

skip_on_upstream_skip=False

flag when you create the task

Thanks for the help @Chris White. That seems to work now.

I think what you're looking for is

merge

, which can be used to merge two branches of a flow back into one (the first executed task passed to merge is its result).

from prefect.tasks.control_flow import merge

with Flow('MyFlow') as flow:
    do_thing = Parameter('do_think', default=False, required=False)
    
    data = get_data()
    
    with case(do_thing, True):
        data2 = do_thing_task(data)
    
    # `data2` if `data2` wasn't skipped, otherwise `data`
    data = merge(data2, data)
    
    other_data = do_another_thing(data)