In Prefect 2, tasks fail if their Python function raises an exception. You can enable retries by passing

retries

and

retry_delay_seconds

parameters to your task. If the task fails, Prefect will retry it up to

retries

times, waiting

retry_delay_seconds

seconds between each attempt. If the task fails on the final retry, Prefect marks the task as crashed if the task raised an exception or failed if it returned a string. For handling failures, Prefect captures all exceptions to report states to the orchestrator. When calling flows or tasks, the exceptions are raised as in normal Python. You can use a try/except block to handle these exceptions. If you would prefer to check for a failed task without using try/except, there are other ways as well. Here's an example:

python
from prefect import flow, task

@task
def my_task():
    raise ValueError()

@flow
def my_flow():
    try:
        my_task()
    except ValueError:
        print("Oh no! The task failed.")
    return True

my_flow()

For more detailed information, you can refer to the Prefect documentation here and here.