How can I set a flow’s

on_failure

and

on_cancellation

dynamically via the SDK?

Flow.with_options(on_failure=[<foo>], on_cancellation=[<bar>])

could work? Depends on what you're trying to do I guess.

I’m using the decorator though, so I was hoping to have some access internally. Specifically I have some cleanup function defined within the flow, that I’d like to be called with parameters also derived within the flow, i.e.

def cleanup(foo, bar):
    # do stuff with foo and bar

@prefect.flow(name=...)
def the_flow():
    # do stuff
    a = ...
    b = ...
    # here I'd like to be able to do smth like the following
    this_flow.on_cancellation = lambda: cleanup(a, b)

Oh, then I'm guessing you're trying to clean up some resources on failures?

Yes that’s right

Rather than passing explicit Things To Clean up, you could just have a module-level object storing the stuff to clean up in its state and a cleanup() function that uses it wired to the hooks in the usual way in the decorator

So interacting with a global variable to maintain state for cleaning up? That would work, feels a bit off, but that sounds like the best way forward now. The

on_cancellation

argument is new, and we’ve pinned prefect to some previous version. In older versions, would

on_failure

also trigger for cancellation and crashes?

No idea on the latter; on the former - this is already a Python pattern,

contextlib.ExitStack

(Also, would

def cleanup

in your example above require the additional arguments Prefect expects? Or are those optional?)

It's mockup code to convey the general idea more explicitly, it would probably require these (and other stuff like removing itself, possibly use func/arg tuples, etc.) :^)

Makes sense, just wanted to confirm 🙂 Thanks Jan, I’ll look into it!

nw! 🙂

Aw, it seems

on_failure

was literally only for the failed state. Would you know if these server-side triggers also result in some internal exceptions that can be caught?