Is there a way to override a flow run’s state? Context - I’m using

prefect-dbt

to run dbt via the command line, and even though some models are resulting in errors, the flow run transitions to a

completed

state. My thought is to expand the flow logic to parse the dbt output logs to determine if there were any errors, and change the flow run state if so. Not sure if that’s the right way to be thinking about this issue though.

from prefect import flow
from prefect.states import Completed, Failed

@flow
def my_flow():
    if some_condition:
        return Completed(message="Success!")
    else:
        return Failed(message="Failure!")

There may be better way to do it, but this is what I've done in the past