https://prefect.io logo
Title
r

Robert Banick

05/04/2023, 11:13 PM
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!
a

Abhinav Chordia

05/05/2023, 2:29 AM
Can’t you just place these under a tmpdir?
j

justabill

05/05/2023, 12:10 PM
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.
r

Robert Banick

05/05/2023, 3:39 PM
@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.
j

justabill

05/08/2023, 1:24 AM
Yes, you could reference with pretty much anything in the runtime context, but flow name makes sense to me.