"With Case" statement + Results from task


I was using “with case” to build a conditional flow based on return values from tasks, but (per my understanding) this works only with bool values, so how can I get any other types? e.g. strings or more a complex object with specific properties.

In summary I need a task that executes a POST request, and based on the body response return an ID (hence ‘conditional flow’ concept + with case got into picture)

Later this ID will be used next in a subsequent task, that will create another request.

Below code with simple if/else statements that shows somehow what I need

Thank you!

def startTask(flowName):
    print("STARTING FLOW " + flowName)

def SubmitRequest(payload):
    nr = random() # -> RequestID(payload) this would call an API 
    print(f"Number: {nr}")
    if nr > 0.5:
        return "123456" #THIS IS ID1
        return ""

def SubmitRequest2(payload):
    nr = random() # -> RequestID(payload) this would call ANOTHER API 
    print(f"Number: {nr}")
    if nr > 0.5:
        return "789012" #THIS IS ID2
        return ""

with Flow("MAIN_SDM_FLOW",) as parent_flow:
    #First task in flow
    id1 = SubmitRequest("")

    if id1 == "":
        raise ValueError("Empty")
        id2 = SubmitRequest2("")

1 Like

Welcome to Discourse!

Before we dive deeper here, where are you in your Prefect adoption? Are you new to Prefect or using it for some time already? If you have the option to choose Prefect 2.0, your use case is much easier to solve:

LMK and this way I can give a better answer

We have are evaluating Prefect as a workflow tool and need to run a simple proof of concept in a production ready environment, thus 1.0 was chosen.

Just FYI: GA of Prefect 2.0 is just around the corner - the target estimate for that is July

It totally works with strings, it depends on what you are trying to do - can you share your attempt of using strings in the case statement and what didn’t work there?

Thanks! I didn’t know it would compare strings too.

So now I ran code below and I am returning “123456” from submit_request(), so “empty_value” will always be false but the 2nd request (close_request) doesn’t happen. What would be the best way to run conditional flows like this?

with Flow("MAIN_SDM_FLOW") as flow:
    empty_value = False
    id1 = submit_request()
    with case(id1, ""):
        empty_value = True
    if not empty_value:
        id2 = close_request("")  

maybe I should have clarified that but the reason why case exists is that in Prefect 1.0, you can’t run any Python code in your Flow other than just calling your tasks and building a DAG - so the if statement should also be replaced with another case statement