Does anyone know the preferred way to get the state of a Flow after execution? I tried to write a terminal_state_handler but cannot find how to import the 'State' module. Is this even the correct approach or is there another way that is better?

Can you explain your use case first? What would you want to accomplish with such a state handler?

@Anna Geller Thanks. My use case is that I want to capture whether a flow has succeeded or failed in our internal monitoring tool For this, I would need to determine the SUCCESS or FAIL status of the workflow. I think it means I need to define a terminal_state_handler and walk the task-states to see if there is a case where 'is_failed()' is true. My question was really about whether this was the best/preferred way of doing this.

glad you explained your use case more! so your end goal would be to get an alert when a flow run fails? are you on Prefect Cloud or Server?

I'm am developing on server but will run production in cloud.

what's your alerting solution? Slack, Teams, Email, Pagerduty?

I don't know if it is a case of alert. I see it as more of post workflow processing.

Just use a vanilla state handler attached to the flow:

def state_handler(obj: Flow, old_state: State, new_state: State) -> Optional[State]:
    """
    if new_state.is_finished():
        if new_state.is_failed():
            ....
        if new_state.is_successful():
            ....
    return new_state

Where is the State imported from?

tnx

You can find that here

as Kevin mentioned, there are 2 options for alerting on failed flow run: one would be via a flow-level state handler (make sure to attach it to your Flow object, rather than to a task) as above and another via a Cloud Hook https://docs.prefect.io/orchestration/concepts/cloud_hooks.html#cloud-hooks

Thanks a bunch. I'll look into them.