"With Case" statement + Results from task

Archive
,
1

Hi

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!

@task(log_stdout=True)
def startTask(flowName):
    print("STARTING FLOW " + flowName)

@task
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
    else:
        return ""

@task
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
    else:
        return ""


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

    if id1 == "":
        print("Error")
        raise ValueError("Empty")
    else:
        id2 = SubmitRequest2("")
1 Like
2

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:

https://discourse.prefect.io/t/should-i-start-with-prefect-2-0-orion-skipping-prefect-1-0/544

LMK and this way I can give a better answer

3

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.

4

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

5

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?

6

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("")
7

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