I have a state handler that checks for is_failed() and sets the state to Skipped if that is the case. This is a requirement. Something like this

def handler(task, old_state, new_state):
    if new_state.is_failed():
        // do stuff here
        return Skipped()
    return new_state

the consequence of this is that the flow finishes with a SUCCESS state, even though a task actually failed, but I need to show the flow as FAILED. To solve this, I created a new flow that calls a different state handler and checks its state. If the state is skipped, I return Failed() from the handler. Something like this:

@task(name="final check", state_handlers=[final_check_handler])
def final_check():
    pass

and the state handler

def final_check_handler(task, old_state, new_state):
    if new_state.is_skipped():
        return Failed()

This accomplishes what I want, but surely there has to be a better way.

Hey @David Jenkins, won’t the Failed propagate and downstream tasks won’t run? Wondering why you return Skipped() instead of Failed() there? I guess you might have downstream state handlers that check if something is Skipped(). In this case, I am not sure there is a good way to handle it, except that you can put that final state handler at the Flow level. We have a terminal_state_handler that will run after all of the tasks.

If I don't reset it to skipped, the fail will propagate, true. However, my requirement is that the subsequent tasks are skipped, not failed. This will propagate the skip and the flow will finish in a SUCCESS state. I need a way to show the flow finished in a FAILED state.

I understand. I think what you have there then is pretty much the most that can be done (except doing it at the flow level instead)

yeah. I do have it at the flow level. It just seems kinda kludgy and was hoping there is a more elegant way to do it.