python 43 lines · 8 steps

Chaining Celery tasks in Django

A report is generated, then emailed, then error-handled by linking Celery tasks into one asynchronous workflow.

Explained by highlit
Django Intermediate asynchronous tasks task chaining retries callbacks background jobs
1from celery import shared_task, signature
2from django.core.mail import send_mail
3 
4from .models import Report, ReportExport
5 
6 
7@shared_task(bind=True, max_retries=3, default_retry_delay=30)
8def generate_report(self, report_id):
9 report = Report.objects.get(pk=report_id)
10 try:
11 export = report.build_export()
12 except Report.DataUnavailable as exc:
13 raise self.retry(exc=exc)
14 
15 report.mark_generated()
16 return export.pk
17 
18 
19@shared_task
20def notify_report_ready(export_id):
21 export = ReportExport.objects.select_related("report__owner").get(pk=export_id)
22 owner = export.report.owner
23 send_mail(
24 subject=f"Your report '{export.report.title}' is ready",
25 message=f"Download it here: {export.download_url()}",
26 from_email=None,
27 recipient_list=[owner.email],
28 )
29 export.mark_notified()
30 
31 
32def schedule_report(report_id):
33 workflow = generate_report.signature(
34 (report_id,),
35 link=notify_report_ready.s(),
36 link_error=flag_report_failure.s(report_id),
37 )
38 return workflow.apply_async()
39 
40 
41@shared_task
42def flag_report_failure(request, exc, traceback, report_id):
43 Report.objects.filter(pk=report_id).update(status=Report.Status.FAILED)
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Linking tasks with callbacks lets you build a pipeline where each stage triggers the next without blocking.
  2. 2Binding a task with `bind=True` exposes `self.retry`, turning transient failures into automatic retries instead of hard errors.
  3. 3Separating success (`link`) and failure (`link_error`) callbacks keeps the happy path and error handling cleanly decoupled.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
python 
import re
from functools import total_ordering
from typing import Optional
 

Parsing and comparing semantic versions

regex operator-overloading sorting
Intermediate 7 steps
python 
from typing import Any, Sequence, Mapping
 
 
def render_markdown_table(

Rendering an aligned Markdown table in Python

string formatting data transformation closures
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Chaining Celery tasks in Django — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code