Handling Failure

Looking for the latest Prefect 2 release? Prefect 2 and Prefect Cloud 2 have been released for General Availability. See https://docs.prefect.io/ for details.

Follow along in the Terminal

cd examples/tutorial
python 04_handle_failures.py

If at first you don't succeed...

Now that we have a working ETL flow let's take further steps to ensure its robustness. The extract_* tasks are making web requests to external APIs in order to fetch the data. What if the API is unavailable for a short period? Or if a single request times out for unknown reasons? Prefect Tasks can be retried on failure; let's add this to our extract_* tasks:

from datetime import timedelta
import aircraftlib as aclib
from prefect import task, Flow, Parameter


@task(max_retries=3, retry_delay=timedelta(seconds=10))
def extract_reference_data():
    # same as before ...
    ...


@task(max_retries=3, retry_delay=timedelta(seconds=10))
def extract_live_data(airport, radius, ref_data):
    # same as before ...
    ...

This is a simple measure that helps our Flow gracefully handle transient errors in only the tasks we specify. Now if there are any failed web requests, a maximum of 3 attempts will be made, waiting 10 seconds between each attempt.

More Ways to Handle Failures

There are other mechanisms Prefect provides to enable specialized behavior around failures:

• Task Triggers: selectively execute Tasks based on the states from upstream Task runs.
• State Handlers: provide a Python function that is invoked whenever a Flow or Task changes state - see all the things!
• Notifications: Get Slack notifications upon state changes of interest or use the EmailTask in combination with Task Triggers.

Up Next!

Schedule our Flow to run periodically or on a custom schedule.