https://prefect.io logo
Title
r

Rob McInerney

05/23/2022, 4:36 PM
Hi! I’m trying to configure reference tasks for a flow with case statements and having some difficulty. Despite tasks failing, the flow is still succeeding. What am I doing wrong? Code follows the following example:
with Flow(FLOW_NAME, storage=set_storage(FLOW_NAME), run_config=set_run_config(),) as flow:
    dataset = Parameter("dataset", default={})
    
    dataset_is_valid = check_dataset_is_valid(dataset)
    
    with case(dataset_is_valid, True):
       
        access_token = get_access_token()
        response = refresh_dataset(dataset, access_token)

        result = check_result(dataset, response, access_token)
        flow.set_reference_tasks([response, result])
    
    with case(dataset_is_valid, False):
        flow.set_reference_tasks([dataset_is_valid])
Both refresh_dataset and check_result are failing at the moment, but the flow is succeeding
k

Kevin Kho

05/23/2022, 4:40 PM
I don’t think you can do it this way.
case
is a task but the
flow.set_reference_tasks
is not. I am not even sure if it runs or not. I think you can do:
flow.set_reference_tasks([response, result, dataset_is_valid])
at the end of the Flow block. Anyway, the other stuff will be SKIPPED which would be considered a success.
r

Rob McInerney

05/23/2022, 4:41 PM
Ok understood, I’ll give that a go!
Tested and that works fine. Thanks for the help
👍 1