@Marvin How do I return a Failed state from a flow in Prefect 2?

Edited by @Nate: Ah, the sweet scent of failure. In Prefect 2, the final state of a flow is determined by its return value. Here are some rules that apply: - If an exception is raised directly in the flow function, the flow run is marked as failed. - If the flow does not return a value (or returns None), its state is determined by the states of all of the tasks and subflows within it. - If any task run or subflow run failed, then the final flow run state is marked as FAILED. To return a Failed state from a flow, you can raise an exception within the flow function. Here's an example:

python
from prefect import flow

@flow
def always_fails_flow():
    raise ValueError("This flow immediately fails")

always_fails_flow()

Running this flow will produce a result with a Failed state. You can also return a manually created state, like this:

python
from prefect import task, flow
from prefect.states import Failed

@task
def always_fails_task():
    raise ValueError("I fail successfully")

@task
def always_succeeds_task():
    print("I'm successful")

@flow
def mixed_flow():
    always_fails_task()
    always_succeeds_task()
    return Failed("Flow failed due to a failed task")

mixed_flow()

In this example, the flow returns a manually created Failed state, which determines the final state of the flow run. So, embrace the darkness and enjoy your failures. After all, humans learn from them, right?

Thanks but I think "from prefect.engine.state" is for Prefect 1

you're right, the correct import is

from prefect.states import Failed

- updating Marvin's answer