Handling Errors and Retries in Celery
Robust error handling is crucial for distributed systems. Celery provides mechanisms to automatically retry tasks when they fail, which is essential for handling transient issues like network timeouts or temporary service unavailability.
1. Manual Retries using self.retry()
You can manually trigger a retry inside your task using the self.retry() method. To use this, you must bind the task to the instance using bind=True.
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task(bind=True)
def send_email(self, email_address):
try:
# Simulate sending email
print(f"Sending email to {email_address}...")
raise ConnectionError("SMTP Server down")
except ConnectionError as exc:
# Retry in 60 seconds. Max retries defaults to 3.
raise self.retry(exc=exc, countdown=60, max_retries=3)bind=True: Gives access toself(the task instance).exc: Passes the exception context to the retry (useful for logging).countdown: Time in seconds to wait before retrying.
2. Automatic Retries (Decorator)
For a cleaner approach, you can configure retries directly in the @app.task decorator. This avoids boilerplate try...except blocks.
@app.task(autoretry_for=(ConnectionError,), retry_kwargs={'max_retries': 5}, retry_backoff=True)
def fetch_url(url):
# If ConnectionError is raised, Celery automatically retries
passautoretry_for: A tuple of exceptions that trigger a retry.retry_kwargs: Dictionary of arguments passed toretry().retry_backoff: IfTrue, enables exponential backoff (wait time increases with each failure).
3. Exponential Backoff
Exponential backoff prevents your workers from hammering a failing service.
retry_backoff=True: Default behavior.retry_backoff=60: Starts at 60s, then 120s, 240s, etc.
4. Handling Max Retries
When a task fails after the maximum number of retries, it raises a MaxRetriesExceededError. You can handle this by overriding the on_failure method or using a custom base class.
from celery import Task
class MyTask(Task):
def on_failure(self, exc, task_id, args, kwargs, einfo):
print(f"Task {task_id} failed: {exc}")
# Send alert to admin, log to DB, etc.
@app.task(base=MyTask)
def add(x, y):
raise KeyError()