Hi I m currently migrating to prefect 2 from 1 there are a f Prefect Community #ask-community

Hi, I'm currently migrating to prefect 2 from 1, ...

Ed Burroughes

03/27/2023, 3:16 PM

Hi, I'm currently migrating to prefect 2 from 1, there are a few things I'm unable to find equivalent task params for in prefect 2, see below examples for more details: 1. Getting the results to persist using PrefectResult():

Copy code

@task(result=PrefectResult())
def return_results(table_name):
    return {"table_name": table_name)

2. Running the task no matter the upstream state using

trigger=all_finished:

Copy code

@task(result=PrefectResult(), trigger=all_finished)
def return_results(table_name):
    return {"table_name": table_name)

Can anyone point me in the right direction to the equivalent Prefect 2 features, thank you 🙂

Zanie

03/27/2023, 3:18 PM

@task(persist_result=True)

and

my_task(x=1, y=allow_failure(upstream_task))

Zanie

03/27/2023, 3:18 PM

https://docs.prefect.io/concepts/results/#persisting-results

Ed Burroughes

03/27/2023, 3:21 PM

Thank you very much 👍

Ed Burroughes

03/28/2023, 1:09 PM

So using the allow_failure approach from above, I have multiple tasks and it ends up looking like the below:

Copy code

_return_results = return_results(
        table_name,
        trigger_s3_key,
        cleaned_aux_file_path,
        str(flow_run_obj.flow_run.id),
        wait_for=[
            allow_failure(csv_file),
            allow_failure(csv_dialect),
            allow_failure(table_obj),
            allow_failure(_drop_table),
            allow_failure(_create_table),
            allow_failure(_load_data),
            allow_failure(_shrink_columns),
])

Is there a nicer way to allow_failure on all upstream tasks in prefect 2?

Ed Burroughes

03/28/2023, 1:18 PM

FYI on the above @Zanie

Zanie

03/30/2023, 4:04 PM

Why not something more Pythonic?

Zanie

03/30/2023, 4:04 PM

Copy code

from prefect import flow, task, allow_failure


@flow
def my_flow():
    x = fail_task.submit()
    y = fail_task.submit()
    my_task(wait_for=[allow_failure(u) for u in (x, y)])
    return True


@task(log_prints=True)
def my_task():
    print("hi")


@task
def fail_task():
    raise ValueError()


my_flow()

Open in Slack

Previous Next