PrefectV2 appears to prefer throwing an exception ...
# prefect-community
r
PrefectV2 appears to prefer throwing an exception to abort a flow early based on a condition (https://discourse.prefect.io/t/how-can-i-stop-a-flow-run-execution-based-on-a-condition/105). However, this results in failed run. Is there any way to end a flow early conditionally without a failed state. For example, we would like to run a daily dbt rollup when our source table has 24 hours of data, if not we should skip the dbt task. In this case, skipping the task is expected and show not results in a failed state.
looks like a simple if statement will work here -

https://www.youtube.com/watch?v=q1-bc_vJZbE

🙌 1
a
Is there any way to end a flow early conditionally without a failed state.
yes, 100% you can e.g. return a state early - a silly flow example that shows how you can return states:
Copy code
from prefect import task, flow
from prefect.orion.schemas.states import Completed, Failed


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


@task
def always_succeeds_task():
    print("I'm fail safe!")
    return "success"


@flow
def return_state_manually():
    x = always_fails_task.submit()  # .result(raise_on_failure=False)
    y = always_succeeds_task.submit()
    if y.result() == "success":
        return Completed(message="I am happy with this result")
    else:
        return Failed(message="How did this happen!?")


if __name__ == "__main__":
    return_state_manually()
👍 1