Hello, I am looking into Prefect 2.0 and am wondering how we might implement a cleanup function - in prefect 1 from what I have seen we could use triggers "All Finished" when defining the task - I cant see the equivalent in V2, any suggestions?

I think wrapping a

finally

statement works

from prefect import flow, task

@task
def do_something() -> bool:
    return True


@task
def do_something_bad():
    raise RuntimeError("Oops")


def cleanup():
    print("cleaning up")


@flow
def test():
    try:
        do_something()
        do_something()
        do_something_bad()
    finally:
        cleanup()

test()

Andrew’s suggestion is interesting. I was just thinking:

@flow
def test():
    a = task_one(raise_on_exception=False)
    b = task_two(raise_on_exception=False)
    c = task_three(raise_on_exception=False)
    cleanup(wait_for=[a,b,c])

I was wondering if cleanup could run prematurely with the first approach?

That

ReservedArgumentrError

is on the function that is decorated during runtime because it’s reserved for the

call

from my understanding

you’re right; it did run prematurely; I wonder if that should be the expected behavior or not

It might be premature because the call just creates the Future and is technically done executing right? You’d need to call .wait() or .result()

yep that makes sense. since finally statements are usually called for cleanup, I wonder if prefect should raise a warning? but I guess if something is passed to it, it’ll wait properly

Kevin, it doesnt seem to like raise_on_exception - for the happy path the cleanup gets called fine, but the following is failing

import requests
from prefect import flow, task
from time import sleep

@task(retries=3,retry_delay_seconds=3)
def do_something():
    print("This should run in parallel?")
    sleep(2)

@task(retries=1,retry_delay_seconds=3)
def call_api(url):
    response = requests.get(url)
    print(response.status_code)
    sleep(2)
    return response.json()

@task(retries=3,retry_delay_seconds=3)
def cleanup():
    print("Got the cleanup call!")

@flow
def api_flow(url):
    a = do_something()
    b = call_api(url, raise_on_exception=False)
    cleanup(wait_for=[a,b])


#api_flow("<http://time.jsontest.com/>") #happy
api_flow("doh")

I think my syntax was wrong one sec