Something I’m not quite sure how to do in `prefect...
# prefect-community
b
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:
Copy code
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
Copy code
def run_flow_steps(x, do_optional_step):
    y = x + 1
    if do_optional_step:
        y = 2*y
    return y
👀 1
n
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:
Copy code
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()
Which would log
4
when
do_optional_step
is
False
, or
2
when it's
True
b
Ah gotcha. That makes sense. So I basically just need a passthrough task for one of the control flow branches
Thanks @nicholas!
n
Happy to help!