a

    Andrew Nichol

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

    Kevin Kho

    1 year ago
    Hi @Andrew Nichol! Have you checked out the Automations tab in the UI?
    a

    Andrew Nichol

    1 year ago
    hmm looks like that could be helpful but id like the state conditions and actions all to be defined in code
    Kevin Kho

    Kevin Kho

    1 year ago
    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 state handler and check for the cancelled state
    a

    Andrew Nichol

    1 year ago
    that looks like it should work - will test and return. thanks @Kevin Kho!
    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
    here is some example code
    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)
    Kevin Kho

    Kevin Kho

    1 year ago
    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!)
    a

    Andrew Nichol

    1 year ago
    okay thanks much for your help