Task retry and error handling in Django

Introduction

Sometimes tasks fail due to temporary problems. Retry and error handling help your app try again or manage errors smoothly without crashing.

When sending emails that might fail due to network issues.

When calling external APIs that can be temporarily down.

When processing background jobs that depend on unstable resources.

When you want to log errors and keep your app running safely.

Syntax

Django

from celery import shared_task
from celery.exceptions import MaxRetriesExceededError

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def my_task(self, arg):
    try:
        # Your task code here
        pass
    except Exception as exc:
        raise self.retry(exc=exc)

@shared_task(bind=True) lets the task access itself to retry.

max_retries sets how many times to retry before giving up.

Examples

This retries sending email after 30 seconds if it fails.

Django

from celery import shared_task

@shared_task(bind=True)
def send_email(self, email):
    try:
        # pretend to send email
        if not email:
            raise ValueError("No email provided")
    except Exception as exc:
        raise self.retry(exc=exc, countdown=30)

This retries fetching data up to 5 times, waiting 2 minutes between tries.

Django

from celery import shared_task

@shared_task(bind=True, max_retries=5, default_retry_delay=120)
def fetch_data(self, url):
    try:
        # pretend to fetch data
        if url == "bad_url":
            raise ConnectionError("Failed to connect")
    except Exception as exc:
        raise self.retry(exc=exc)

Sample Program

This task divides two numbers. If dividing by zero, it retries twice with 10 seconds delay. If still fails, it stops and prints an error.

Django

from celery import shared_task

@shared_task(bind=True, max_retries=2, default_retry_delay=10)
def divide_numbers(self, a, b):
    try:
        result = a / b
        return result
    except ZeroDivisionError as exc:
        print(f"Error: {exc}. Retrying...")
        raise self.retry(exc=exc)

# Simulate calling the task
if __name__ == "__main__":
    task = divide_numbers
    try:
        print(task(10, 0))
    except Exception as e:
        print(f"Task failed after retries: {e}")

Important Notes

Retries help with temporary problems but don't fix permanent errors.

Use max_retries to avoid infinite retry loops.

Logging errors inside except blocks helps track issues.

Summary

Task retry lets your app try again when something goes wrong temporarily.

Use bind=True to access retry methods inside tasks.

Set limits on retries and delays to control task behavior.