Flow with interchangeable Parameters

[gryBox]

Hi - I am trying to write a flow with interchangeable parameters. The flow executes \with no error thrown. But:

• The terminal task is always skipped when both parameters are passed

• The flow does not accept passing one or the other Parameter.

Is this an anti pattern or bug?

Code

from prefect import task, Flow, Parameter
from prefect.tasks.control_flow import  case, ifelse, merge
from prefect.engine.executors import LocalExecutor

@task
def get_greeting(name, id_dict=None):
    if id_dict:
        greeting = f"Hello {id_dict['some_id']} {name}"
    else:
        greeting =  f"Hello {name}"
    return greeting


with Flow("Confused-Get") as flow:

    name = Parameter("name", default=None) 
    id_dict = Parameter("id_dict", default=None)
    #########

    # Handle missing name
    with case(name.is_equal(None), True):
        name = id_dict["name"]

    # 1. Do something
    greeting = get_greeting(
        name,
        id_dict
    )

id_dict = {
    "name": "Sam",
    "some_id": "Polka-dot"
}

flow.visualize()
flow_state = flow.run(
    parameters=dict(
        name=id_dict["name"],
        id_dict=id_dict
    ),
    executor= LocalExecutor()   
)

flow.visualize(flow_state=flow_state)

Thanks

[cicdw]

Hi @gryBox - remember that building a Flow represents building a deferred computational graph. So as much as it looks and feels like "normal" python code, every line is run. In particular,

    # Handle missing name
    with case(name.is_equal(None), True):
        name = id_dict["name"]

overrides your original name = Parameter("name", default=None) task with a new GetItem task. You should instead do something like (untested code):

    # Handle missing name
    with case(name.is_equal(None), True):
        greeting1 = get_greeting(
        id_dict["name"],
        id_dict
    )
   with case(name.is_equal(None), False):
        greeting2 = get_greeting(
        name,
        id_dict
    )