Something I’m not quite sure how to do in

prefect

is use control flow to have an optional task inline in a sequence of tasks. Something like:

with Flow('flow') as flow:
    do_optional_step = Parameter('do_optional_step')
    
    x = Parameter('x')
    y = x + 1
    y = ifelse(do_optional_step, 2*y, y)

This doesn’t work as intended, but is what I’d like to be able to do. If I were doing this as a pure python function, it would be

def run_flow_steps(x, do_optional_step):
    y = x + 1
    if do_optional_step:
        y = 2*y
    return y

Hi @Ben Fogelson! I think the problem in your code there is that the

ifelse

task takes 2 tasks as parameters, rather than 2 static values. So you could opt to do something like this:

import prefect
from prefect import task, Flow, Parameter
from prefect.tasks.control_flow.conditional import ifelse


@task
def add(x):
    return x + 1


@task
def keep_same(x):
    return x


@task
def double(x):
    return x * 2


@task
def print_task(input):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(f"Result: {input}")


with Flow("hello-flow") as flow:
    do_optional_step = False
    x = 1

    y = add(x=x)

    z = ifelse(do_optional_step, keep_same(y), double(y))

    print_task(z)


flow.run()

Ah gotcha. That makes sense. So I basically just need a passthrough task for one of the control flow branches

Happy to help!