Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

howdy
id like to have an action in my tasks/flows happen upon cancellation from the prefect ui - is this doable?

Hi <@U02216RKYN5>! Have you checked out the Automations tab in the UI?

hmm looks like that could be helpful but id like the state conditions and actions all to be defined in code

~Let me check with the team if a terminal state handler will still run upon flow cancelletation~

You can’t do it on the task level but you can do it with the <https://docs.prefect.io/core/concepts/flows.html#state-handlers|state handler> and check for the cancelled state

that looks like it should work - will test and return. thanks <@U01QEJ9PP53>!

so it appears that using state_handlers at the flow level works for this whereas at the task level it does not

however upon cancellation i want the flow to do something based on some variables which can be found inside of tasks

the easiest way i found to do this is with global variables

ideally i could avoid that but for now it works

```def run_on_cancel(task, old_state, new_state):
    global some_global_variable
    if isinstance(new_state, prefect_state.Cancelled):
        logger = prefect_context.get("logger")
        <http://logger.info|logger.info>(F"Running CANCELATION: {some_global_variable}")
        raise Exception(f"exiting because of Cancelled state: {some_global_variable}")
    return new_state

@task
def supposed_to_succeed():
    global some_global_variable
    some_global_variable = "beaver"
    logger = prefect_context.get("logger")
    <http://logger.info|logger.info>("Running success")
    sleep(15)
    return True

with Flow(
    "failure-flow",
    executor=LocalDaskExecutor(),
    result=PrefectResult(),
    state_handlers=[run_on_cancel]
) as failure_flow:
    some_global_variable = None
    succeed = supposed_to_succeed()
    maybe_failure = supposed_to_fail(succeed)
    last(maybe_failure)```

This is right. You can’t do this on a task level unfortunately and there’s no way for the state handler to be able to get variables from the task other than the way you did I think. (First time I’ve seen it actually!)