Hi Prefect team and especially @justabill, I have a question about the capabilities of the

on_crashed

parameter just released (!). Is is possible to pass in additional arguments to function so invoked, using objects generated during the course of a flow? Like so:

def cleanup(path_name: str):
    shutil.rmtree(path_name)

def crash_hook(path_name, flow, flow_run, state):
    cleanup(path_name)

@flow(on_crashed=[crash_hook])
def my_flow():
    # stuff
    path_name = other_function()

My use case is wiping very large amounts of data downloaded for temporary processing as a part of my ETLs. Leaving this lying around on temporary storage is expensive and best avoided. The path name for the directory to wipe depends on the ETL so invoked and so a static function won’t cut it; I need to vary the path I apply

shutil.rmtree

to. Thanks for pushing this awesome feature, it directly responds to a problem I was just about to address. Very cool!

Can’t you just place these under a tmpdir?

Interesting use case. I definitely understand the need. Unfortunately, the feature doesn't support passing additional arguments as you've done above. Presently, the best way to done that would be to set the

path_name

as a string block so it can be retrieved.

@justabill OK thanks for clarifying. The string block idea is interesting. I suppose I could reference the correct block using the flow name? That’s the only route I can think of.

Yes, you could reference with pretty much anything in the runtime context, but flow name makes sense to me.