Exécuter des tâches en arrière-plan avec Celery et Redis dans Django
Lorsqu’on commence à développer une application Django, beaucoup de choses semblent simples. On reçoit une requête HTTP, on exécute une fonction Python, on interroge la base de données, on construit une réponse et on renvoie le tout au navigateur. Cette architecture fonctionne remarquablement bien pour un grand nombre de fonctionnalités classiques. Le problème commence généralement lorsque certaines opérations prennent plus de temps que prévu. Envoyer plusieurs centaines d’e-mails, générer un rapport PDF, redimensionner plusieurs images, analyser un fichier importé, appeler une API externe, traiter des milliers d’enregistrements, recalculer des statistiques ou lancer une opération de nettoyage ne sont pas des tâches que l’on souhaite laisser bloquer une requête HTTP.
C’est exactement dans ce contexte que les tâches exécutées en arrière-plan deviennent intéressantes. Au lieu de demander au serveur web d’accomplir immédiatement une opération potentiellement longue, on lui demande de placer cette opération dans une file d’attente. Un autre processus, appelé worker, récupère ensuite la tâche et l’exécute indépendamment de la requête initiale. L’utilisateur peut alors obtenir immédiatement une réponse, pendant que le traitement continue en arrière-plan.
Dans l’écosystème Python, Celery est l’un des outils les plus connus pour construire ce type d’architecture. Redis, de son côté, constitue une solution très pratique pour transporter les messages entre Django et les workers et peut également être utilisé comme backend de résultats. La documentation actuelle de Celery indique que la branche stable 5.6 prend en charge Redis comme broker et backend, et documente explicitement son intégration avec Django.
Il faut cependant éviter de considérer Celery comme une simple commande magique du genre task.delay(). En production, Celery implique une vraie réflexion autour des files, des retries, de l’idempotence, des timeouts, des erreurs, de la surveillance, de la consommation mémoire, des tâches périodiques, des transactions Django et de la façon dont les workers sont lancés et arrêtés. Le but de cet article est donc d’aller beaucoup plus loin qu’une simple installation. Nous allons construire mentalement l’architecture, créer un projet Django, connecter Celery à Redis, définir des tâches, les déclencher, récupérer leurs résultats, gérer les erreurs, introduire des retries, planifier des traitements périodiques, créer plusieurs files, superviser les workers et préparer une configuration réellement exploitable en production.
Pourquoi exécuter des tâches en arrière-plan dans Django ?
La première question à se poser n’est pas « comment installer Celery ? », mais plutôt « pourquoi ai-je besoin d’un système de tâches en arrière-plan ? ». Cette distinction est importante, car un système asynchrone ajoute une couche de complexité. Il ne faut donc pas le mettre partout simplement parce que cela paraît moderne.
Imaginez une vue Django qui doit créer un compte utilisateur. L’opération consiste à valider un formulaire, créer une ligne dans la base et retourner une réponse JSON. Si l’ensemble prend 30 ou 50 millisecondes, l’ajout d’une file Celery ne ferait qu’ajouter de la complexité. En revanche, imaginons que cette même création déclenche l’envoi de plusieurs e-mails, la génération d’un document PDF, une synchronisation avec un CRM externe et le traitement d’une image de profil. Dans ce cas, attendre que toutes ces opérations soient terminées avant de répondre au navigateur serait une mauvaise idée.
Une requête HTTP doit idéalement rester prévisible. Plus elle dépend d’appels externes, d’opérations CPU ou d’entrées/sorties longues, plus elle devient fragile. Une API externe peut être momentanément lente. Un fournisseur d’e-mails peut répondre après plusieurs secondes. Une génération de PDF peut prendre beaucoup plus longtemps avec un gros rapport. Une image de 20 Mo peut demander plusieurs secondes de traitement. Et lorsque vous multipliez ces opérations par plusieurs utilisateurs simultanés, le problème ne concerne plus seulement l’expérience utilisateur. Il concerne également les ressources de vos processus web.
L’idée fondamentale consiste alors à séparer la responsabilité de réception de la requête et la responsabilité d’exécution du travail.
On peut représenter le modèle de manière très simple :
Navigateur
|
| HTTP Request
v
Django
|
| enqueue task
v
Redis
|
| task message
v
Celery Worker
|
| execute
v
Database / API / Email / File / Report
Django joue le rôle de producteur de tâches. Redis joue généralement le rôle de broker, c’est-à-dire de système intermédiaire qui transporte les messages. Le worker Celery joue le rôle de consommateur. L’application web n’attend plus forcément la fin de la tâche.
Cette séparation apporte plusieurs bénéfices. Le premier est la réactivité. Le second est la possibilité de mettre plusieurs workers en parallèle. Le troisième est la capacité à répartir les traitements sur plusieurs machines. Le quatrième est la possibilité d’utiliser des politiques de retry pour certaines erreurs transitoires. Le cinquième est une meilleure isolation entre les requêtes web et les traitements lourds.
Mais il y a une nuance essentielle : une tâche en arrière-plan n’est pas automatiquement « plus fiable ». Elle change simplement l’endroit et le moment où l’opération est exécutée. Il faut donc réfléchir à ce qui arrive lorsque le worker tombe, lorsque Redis redémarre, lorsqu’une tâche échoue au milieu de son traitement ou lorsqu’une même tâche est exécutée deux fois.
Comprendre Celery avant de l’installer
Celery est un système distribué de gestion de files de tâches. Le principe est relativement intuitif : on définit une fonction comme tâche, on l’envoie dans une file, puis un worker la récupère et l’exécute.
La fonction suivante est un exemple très simple :
from celery import shared_task
@shared_task
def additionner(a, b):
return a + b
La différence entre cette fonction et une fonction Python classique apparaît lors de son invocation. Une fonction normale serait exécutée immédiatement :
resultat = additionner(10, 20)
Avec Celery, on peut demander à la tâche d’être mise en file :
resultat = additionner.delay(10, 20)
Celery retourne alors un objet de résultat asynchrone plutôt que de vous obliger à attendre l’exécution complète. La documentation de Celery décrit delay() comme un raccourci pratique autour de apply_async(), lequel offre davantage d’options de contrôle.
Le mot « asynchrone » mérite ici une petite précision. Celery ne transforme pas magiquement une fonction Python en fonction non bloquante exécutée dans le même processus. Au contraire, il transmet le travail à un autre processus. Le worker est donc un programme distinct du serveur web.
C’est une distinction fondamentale.
Dans une architecture typique, vous pourrez avoir :
Gunicorn / Django
|
+---- Redis
|
+---- Celery Worker 1
|
+---- Celery Worker 2
|
+---- Celery Beat
Le serveur web traite les requêtes. Les workers traitent les tâches. Celery Beat planifie les tâches périodiques. Redis transporte les messages et peut conserver les résultats si vous l’utilisez comme backend.
Redis comme broker et backend de résultats
Redis est souvent utilisé avec Celery parce qu’il est simple à installer, rapide et polyvalent. Celery documente Redis comme un transport stable, et Redis peut jouer à la fois le rôle de broker et celui de backend de résultats. La documentation souligne néanmoins que Redis fonctionne particulièrement bien pour des messages rapides et de petite taille et qu’il faut prendre en compte la mémoire disponible ainsi que les besoins de persistance lorsque Redis est utilisé pour stocker des résultats.
Il faut distinguer deux rôles.
Le premier est le broker :
Django -> Redis -> Celery Worker
Le deuxième est le result backend :
Celery Worker -> Redis -> Django / client
Le broker permet de transporter les tâches. Le backend de résultats permet de stocker les états et éventuellement les valeurs de retour.
Par exemple, une tâche peut avoir l’état :
PENDING
STARTED
SUCCESS
FAILURE
RETRY
et retourner une valeur :
42
Dans un projet simple, utiliser Redis pour les deux rôles est extrêmement pratique :
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/0"
Cependant, dans un environnement plus important, il peut être judicieux de séparer les bases logiques Redis ou même les infrastructures selon les besoins :
Redis DB 0 -> broker
Redis DB 1 -> result backend
Redis DB 2 -> cache Django
Cela ne signifie pas que des bases Redis logiques isolées constituent une séparation de sécurité ou d’infrastructure complète. En production, une séparation plus claire peut être préférable selon les contraintes de disponibilité, de sécurité et de charge.
Une remarque importante sur Django 6 et le nouveau framework Tasks
Les versions récentes de Django ont introduit une API TASKS native. La documentation Django 6.0 décrit désormais un framework Tasks qui fournit le contrat et l’infrastructure permettant de décrire et de suivre des tâches, tout en laissant l’exécution effective à une infrastructure externe. Les backends intégrés de Django incluent notamment un backend immédiat et un backend dummy.
Cette nouveauté ne rend pas Celery obsolète.
Au contraire, pour les architectures qui ont besoin d’un broker, de workers dédiés, de retries avancés, de routage, de queues spécialisées, de planification et d’un écosystème mature, Celery reste une solution majeure. La documentation actuelle de Celery 5.6 décrit explicitement l’intégration Django et Redis.
Il est donc utile de distinguer les responsabilités. Django propose désormais des abstractions natives pour les tâches. Celery propose un système beaucoup plus riche pour exécuter réellement des tâches distribuées sur des workers.
Installer Redis
Sur Ubuntu ou Debian, vous pouvez installer Redis avec :
sudo apt update
sudo apt install redis-server
Puis vérifier son état :
sudo systemctl status redis-server
Selon votre distribution, le nom du service peut différer. Pour tester que Redis répond :
redis-cli ping
Vous devriez obtenir :
PONG
C’est un petit test, mais il vous évitera de passer vingt minutes à chercher un problème Celery alors que Redis n’est simplement pas démarré.
Dans Docker, on peut faire encore plus simplement :
services:
redis:
image: redis:7
ports:
- "6379:6379"
Puis :
docker compose up -d redis
Vous pouvez vérifier :
docker ps
et ensuite :
docker exec -it <nom_du_conteneur> redis-cli ping
La réponse attendue reste :
PONG
En production, il faut cependant éviter de considérer une image Docker locale comme une stratégie complète d’exploitation. Il faut réfléchir à la persistance, aux sauvegardes selon le rôle de Redis, à l’authentification, à TLS si nécessaire, au réseau, à la mémoire disponible et à la supervision.
Créer le projet Django
Prenons une structure classique :
monprojet/
├── manage.py
├── monprojet/
│ ├── __init__.py
│ ├── settings.py
│ ├── urls.py
│ ├── celery.py
│ ├── asgi.py
│ └── wsgi.py
└── commandes/
├── __init__.py
├── admin.py
├── apps.py
├── models.py
├── tasks.py
├── views.py
└── urls.py
Créons un environnement virtuel :
python -m venv .venv
Sous Linux/macOS :
source .venv/bin/activate
Sous Windows :
.venv\Scripts\activate
Puis installons Django et Celery avec le support Redis :
pip install django celery[redis]
La documentation Celery indique que le support Redis nécessite le paquet Python Redis et recommande l’installation via l’extra celery[redis].
Vous pouvez ensuite vérifier les versions :
python -m django --version
celery --version
Pour les projets modernes en 2026, la documentation officielle disponible indique Celery 5.6 comme branche stable actuelle.
Créer le fichier celery.py
Le fichier celery.py se place généralement à côté de settings.py.
# monprojet/celery.py
import os
from celery import Celery
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "monprojet.settings")
app = Celery("monprojet")
app.config_from_object(
"django.conf:settings",
namespace="CELERY",
)
app.autodiscover_tasks()
Ce petit fichier est très important.
La première ligne :
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "monprojet.settings")
indique à Celery quelle configuration Django utiliser.
Ensuite :
app = Celery("monprojet")
crée l’application Celery.
Puis :
app.config_from_object(
"django.conf:settings",
namespace="CELERY",
)
demande à Celery de lire les paramètres dans les settings Django.
L’utilisation du namespace CELERY est recommandée par la documentation de Celery pour les projets Django, ce qui permet d’éviter de mélanger les paramètres de Celery avec ceux de Django.
Enfin :
app.autodiscover_tasks()
permet à Celery de rechercher les modules tasks.py des applications Django installées.
Importer l’application Celery
Dans :
monprojet/__init__.py
ajoutez :
from .celery import app as celery_app
__all__ = ("celery_app",)
L’objectif est de rendre l’application Celery disponible lorsque Django initialise le projet.
Votre structure devient :
monprojet/
├── __init__.py
├── celery.py
├── settings.py
├── urls.py
├── asgi.py
└── wsgi.py
Configurer Redis dans settings.py
Dans settings.py :
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/0"
Une configuration un peu plus complète pourrait être :
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/1"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "Africa/Casablanca"
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60
La séparation des bases logiques Redis peut aider à organiser le broker et le backend :
Redis DB 0 -> Celery broker
Redis DB 1 -> Celery result backend
Même si cela simplifie la lecture et l’organisation, il faut retenir qu’il s’agit toujours du même serveur Redis. Pour des exigences de haute disponibilité ou d’isolation fortes, cette distinction ne remplace pas une architecture d’infrastructure adaptée.
Créer la première tâche Celery
Dans une application Django :
# commandes/tasks.py
from celery import shared_task
@shared_task
def additionner(a, b):
return a + b
Vous pouvez ensuite ouvrir un shell Django :
python manage.py shell
Puis :
from commandes.tasks import additionner
resultat = additionner.delay(10, 20)
print(resultat.id)
Le champ id est l’identifiant de la tâche.
Par exemple :
7e0f1e4b-4e6f-4c9f-8e74-0b53a5c7fc90
La tâche n’a pas besoin de terminer au moment où delay() retourne.
C’est là qu’intervient le worker.
Lancer le worker Celery
Depuis le dossier contenant manage.py :
celery -A monprojet worker --loglevel=info
Vous devriez voir des informations similaires à :
[tasks]
. commandes.tasks.additionner
[INFO/MainProcess] Connected to redis://127.0.0.1:6379/0
[INFO/MainProcess] celery@machine ready.
La documentation officielle de Celery montre également le modèle classique :
celery -A proj worker -l info
pour démarrer un worker avec une application Django.
À partir de ce moment, si vous appelez :
additionner.delay(10, 20)
le message passe par Redis et le worker le récupère.
C’est souvent le moment où tout devient concret. On voit une ligne apparaître dans le terminal du worker et l’on comprend enfin que Celery n’est pas une extension mystérieuse de Django : c’est un système indépendant qui écoute une file et exécute des fonctions.
Comprendre delay() et apply_async()
delay() est pratique :
additionner.delay(10, 20)
mais il s’agit essentiellement d’un raccourci.
Lorsque vous avez besoin de plus de contrôle :
additionner.apply_async(
args=[10, 20],
)
Vous pouvez ajouter un délai :
additionner.apply_async(
args=[10, 20],
countdown=10,
)
La tâche sera alors planifiée pour être envoyée après le délai indiqué.
Vous pouvez aussi utiliser une date d’exécution :
from datetime import datetime, timedelta, timezone
additionner.apply_async(
args=[10, 20],
eta=datetime.now(timezone.utc) + timedelta(minutes=5),
)
Attention toutefois à ne pas transformer eta ou countdown en substitut systématique d’un véritable planificateur. Pour des tâches périodiques, Celery Beat est généralement plus approprié.
Utiliser un vrai cas Django : envoyer un e-mail
Prenons une situation que l’on rencontre partout.
Après l’inscription d’un utilisateur, vous souhaitez lui envoyer un e-mail de bienvenue.
Sans Celery :
def register(request):
user = create_user(...)
send_welcome_email(user)
return JsonResponse({"success": True})
Le problème apparaît si l’envoi SMTP prend deux ou trois secondes.
Avec Celery :
def register(request):
user = create_user(...)
envoyer_email_bienvenue.delay(user.id)
return JsonResponse({"success": True})
La tâche :
# comptes/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .models import User
@shared_task
def envoyer_email_bienvenue(user_id):
user = User.objects.get(pk=user_id)
send_mail(
subject="Bienvenue !",
message=f"Bonjour {user.get_full_name()}",
from_email="noreply@example.com",
recipient_list=[user.email],
)
Une règle importante apparaît ici : envoyez l’identifiant d’un objet plutôt que l’objet Django lui-même.
Évitez :
envoyer_email_bienvenue.delay(user)
et préférez :
envoyer_email_bienvenue.delay(user.id)
Cela réduit les problèmes de sérialisation et rend l’interface de la tâche plus explicite.
Pourquoi il faut éviter de passer des objets complexes
Celery sérialise les arguments pour les transporter au worker.
Un modèle Django possède beaucoup d’état et peut être associé à des relations, des attributs dynamiques ou des structures qui ne se sérialisent pas proprement.
Cette approche :
@shared_task
def traiter_utilisateur(user):
...
peut sembler naturelle mais elle crée une dépendance plus forte entre le code de l’appelant et la tâche.
Cette version est plus robuste :
@shared_task
def traiter_utilisateur(user_id):
user = User.objects.get(pk=user_id)
...
L’objet est rechargé au moment où la tâche s’exécute.
Cela a également une autre conséquence intéressante : le worker travaille avec l’état actuel de la base au moment de l’exécution.
La relation entre Django, Redis et Celery
Il est utile de visualiser le cycle complet d’une tâche :
1. Requête HTTP
|
v
2. Vue Django
|
| task.delay(...)
v
3. Celery sérialise le message
|
v
4. Redis reçoit le message
|
v
5. Worker Celery consomme
|
v
6. La fonction Python s’exécute
|
v
7. Le résultat est stocké
|
v
8. Redis backend
Imaginez un utilisateur qui clique sur « Générer le rapport ».
Django ne génère pas forcément le rapport. Django répond :
{
"status": "queued"
}
Puis le worker travaille.
Cette séparation est particulièrement intéressante avec une API REST ou une application React / Next.js. Le front-end peut recevoir un task_id :
{
"task_id": "df2e3e44-2bf3-470b-a72d-2dc7ceab9a10"
}
et interroger un endpoint de statut.
Utiliser AsyncResult
Lorsque vous appelez :
result = additionner.delay(10, 20)
vous obtenez un AsyncResult.
Vous pouvez inspecter :
result.id
Puis :
result.state
Par exemple :
PENDING
ou :
SUCCESS
Une fois terminée :
result.get()
peut retourner :
30
Il faut toutefois être prudent avec get(). Dans une requête web, attendre activement la fin d’une tâche revient à réintroduire une partie du problème que Celery devait résoudre.
Mauvaise approche :
result = additionner.delay(10, 20)
value = result.get()
return JsonResponse({"value": value})
Ce modèle transforme votre requête HTTP en attente synchronisée.
Meilleure approche :
result = additionner.delay(10, 20)
return JsonResponse({
"task_id": result.id,
"status": "queued",
})
Puis vous pouvez avoir un endpoint dédié :
from celery.result import AsyncResult
from django.http import JsonResponse
def task_status(request, task_id):
result = AsyncResult(task_id)
return JsonResponse({
"task_id": task_id,
"state": result.state,
})
Construire une API de statut propre
Dans une application réelle, vous pouvez aller plus loin :
from celery.result import AsyncResult
from django.http import JsonResponse
def task_status(request, task_id):
result = AsyncResult(task_id)
data = {
"task_id": task_id,
"state": result.state,
}
if result.successful():
data["result"] = result.result
if result.failed():
data["error"] = str(result.result)
return JsonResponse(data)
Votre front-end peut alors faire :
async function getTaskStatus(taskId) {
const response = await fetch(`/api/tasks/${taskId}/`);
return response.json();
}
Puis interroger périodiquement le serveur.
const interval = setInterval(async () => {
const data = await getTaskStatus(taskId);
if (data.state === "SUCCESS") {
clearInterval(interval);
console.log("Tâche terminée", data.result);
}
if (data.state === "FAILURE") {
clearInterval(interval);
console.error("La tâche a échoué", data.error);
}
}, 2000);
Cela fonctionne, mais dans une application très interactive, vous pourrez aussi choisir une architecture basée sur WebSocket, Server-Sent Events ou notifications push.
Une règle essentielle : une tâche doit être idempotente autant que possible
Le mot « idempotence » peut sembler théorique, mais il correspond à une question très concrète :
Que se passe-t-il si ma tâche est exécutée deux fois ?
Prenons une tâche qui crédite un compte :
@shared_task
def crediter_compte(user_id, montant):
user = User.objects.get(pk=user_id)
user.balance += montant
user.save()
Si la tâche est exécutée deux fois, le compte est crédité deux fois.
C’est dangereux.
À l’inverse, considérons une tâche qui marque une facture comme exportée :
@shared_task
def exporter_facture(invoice_id):
invoice = Invoice.objects.get(pk=invoice_id)
if invoice.exported:
return
exporter(invoice)
invoice.exported = True
invoice.save(update_fields=["exported"])
Cette tâche est beaucoup plus proche d’une opération idempotente.
Dans un système distribué, l’idempotence n’est pas un luxe. C’est une protection contre les redéliveries, les retries, les erreurs de réseau et les redémarrages.
Gérer les erreurs avec autoretry
Certaines erreurs sont temporaires.
Une API externe peut renvoyer :
HTTP 503
Une connexion SMTP peut échouer temporairement.
Redis peut être momentanément indisponible.
Un service tiers peut être ralenti.
Dans ce genre de cas, un retry a du sens.
Celery permet de configurer l’auto-retry :
from celery import shared_task
@shared_task(
autoretry_for=(Exception,),
retry_backoff=True,
retry_kwargs={"max_retries": 5},
)
def appeler_service_externe():
...
Mais attention au :
autoretry_for=(Exception,)
Utiliser Exception sans discernement peut cacher de vrais bugs de programmation.
Une approche plus précise :
class ServiceTemporairementIndisponible(Exception):
pass
Puis :
@shared_task(
autoretry_for=(ServiceTemporairementIndisponible,),
retry_backoff=True,
retry_kwargs={"max_retries": 5},
)
def synchroniser_client(client_id):
...
Le principe est simple : ne retry que les erreurs qui peuvent raisonnablement être corrigées par un nouvel essai.
Utiliser retry() manuellement
Vous pouvez aussi demander explicitement un retry :
from celery import shared_task
@shared_task(bind=True, max_retries=5)
def appeler_api(self):
try:
response = effectuer_appel()
return response
except Exception as exc:
raise self.retry(exc=exc, countdown=60)
Ici :
bind=True
permet à Celery d’injecter l’instance de tâche dans la méthode via self.
Vous pouvez utiliser :
self.retry(...)
pour programmer un nouvel essai.
Une version progressive peut être :
@shared_task(bind=True, max_retries=5)
def appeler_api(self):
try:
return effectuer_appel()
except TimeoutError as exc:
raise self.retry(
exc=exc,
countdown=2 ** self.request.retries,
)
La durée augmente :
Tentative 1 -> 1 seconde
Tentative 2 -> 2 secondes
Tentative 3 -> 4 secondes
Tentative 4 -> 8 secondes
...
En production, il est préférable d’adopter une stratégie de backoff contrôlée plutôt que de bombarder immédiatement un service déjà en difficulté.
Pourquoi il ne faut pas retry toutes les erreurs
Imaginez :
def traiter_commande(order_id):
order = Order.objects.get(pk=order_id)
Si l’objet n’existe pas, répéter la requête cinq fois ne va pas le faire apparaître.
Une erreur de validation :
ValueError
ou un bug de code :
AttributeError
ne doit normalement pas être traité comme une panne temporaire.
La bonne question avant un retry est :
Est-ce que le contexte externe pourrait être différent quelques secondes ou minutes plus tard ?
Si oui, un retry peut être pertinent.
Si non, vous avez probablement besoin d’une correction plutôt que d’une nouvelle tentative.
Ajouter des limites de temps
Une tâche ne doit pas pouvoir rester bloquée indéfiniment.
Celery permet de définir des limites :
CELERY_TASK_TIME_LIMIT = 1800
CELERY_TASK_SOFT_TIME_LIMIT = 1500
Vous pouvez également définir cela au niveau d’une tâche :
@shared_task(
time_limit=300,
soft_time_limit=240,
)
def generer_rapport(report_id):
...
La différence entre limite dure et limite souple est importante.
La limite souple vous donne la possibilité de gérer la situation depuis le code.
Par exemple :
from celery.exceptions import SoftTimeLimitExceeded
from celery import shared_task
@shared_task(soft_time_limit=300)
def traiter_gros_fichier(path):
try:
traiter(path)
except SoftTimeLimitExceeded:
nettoyer_ressources()
raise
L’objectif est de permettre un nettoyage avant l’arrêt forcé.
Les timeouts sont particulièrement utiles lorsqu’une tâche réalise des appels externes ou traite de gros fichiers.
Les queues Celery
Toutes les tâches n’ont pas nécessairement le même profil.
Imaginez votre application avec :
emails
images
rapports
notifications
scraping
Vous pouvez créer des files spécialisées.
Exemple :
@shared_task(queue="emails")
def envoyer_email(user_id):
...
Et :
@shared_task(queue="images")
def traiter_image(image_id):
...
Puis lancer des workers spécialisés :
celery -A monprojet worker -Q emails --loglevel=info
et :
celery -A monprojet worker -Q images --loglevel=info
Cette séparation est extrêmement puissante.
Un traitement d’image très lourd ne devrait pas forcément ralentir le worker qui s’occupe des e-mails transactionnels.
Vous pouvez donc créer une architecture :
Redis
|
+-- emails --> Worker Email
|
+-- images --> Worker Image
|
+-- reports --> Worker Reports
Pourquoi le routage devient important à grande échelle
Au début d’un projet, un seul worker suffit :
celery -A monprojet worker -l info
À mesure que l’application grandit, toutes les tâches finissent par se retrouver dans la même file.
Supposons :
email -> 100 ms
thumbnail -> 10 s
PDF -> 30 s
scraping -> 60 s
Une tâche de scraping peut occuper longtemps un worker.
Si tous les workers sont occupés par des tâches lourdes, un simple e-mail peut attendre alors qu’il ne prendrait que quelques millisecondes à être envoyé.
D’où l’intérêt d’un routage explicite.
Configurer les routes
Dans settings.py :
CELERY_TASK_ROUTES = {
"comptes.tasks.envoyer_email_bienvenue": {
"queue": "emails",
},
"images.tasks.redimensionner_image": {
"queue": "images",
},
"rapports.tasks.generer_rapport": {
"queue": "reports",
},
}
Puis :
celery -A monprojet worker -Q emails -l info
et :
celery -A monprojet worker -Q images -l info
Cette stratégie permet de contrôler plus finement les ressources.
Comprendre la concurrence des workers
Un worker peut exécuter plusieurs tâches selon sa configuration de concurrence.
Exemple :
celery -A monprojet worker --concurrency=4 -l info
Cela signifie que le worker peut utiliser plusieurs processus d’exécution selon le pool configuré.
Cependant, « augmenter le nombre de workers » n’est pas toujours synonyme de « accélérer ».
Si votre tâche est CPU-bound, vous allez rapidement entrer en concurrence pour les CPU.
Si elle est I/O-bound, davantage de concurrence peut parfois améliorer l’utilisation des ressources.
Prenons une tâche qui attend une API externe :
response = requests.get("https://example.com")
Le CPU peut être presque inutilisé pendant l’attente réseau.
À l’inverse, une tâche comme :
for item in massive_dataset:
calcul_complexe(item)
consomme réellement du CPU.
Il faut donc mesurer avant d’ajuster.
Les tâches CPU intensives et les tâches I/O
C’est une distinction essentielle pour dimensionner Celery.
Une tâche I/O-bound :
- API externe
- SMTP
- S3
- téléchargement
- lecture disque
- base de données
passe une partie importante de son temps à attendre.
Une tâche CPU-bound :
- compression
- calcul scientifique
- traitement vidéo
- génération complexe
- analyse de données lourde
consomme le processeur.
Dans le second cas, multiplier aveuglément la concurrence peut saturer la machine sans produire de gain réel.
Pour des traitements particulièrement lourds, une architecture dédiée peut être préférable.
Tâches et transactions Django
Voici un problème classique.
Vous créez un utilisateur dans une transaction :
from django.db import transaction
with transaction.atomic():
user = User.objects.create(...)
envoyer_email_bienvenue.delay(user.id)
Le problème potentiel est que le worker peut démarrer avant que la transaction soit réellement validée.
Le worker reçoit :
user_id = 15
et tente :
User.objects.get(pk=15)
Mais la transaction de la requête web peut ne pas être commitée au moment de la lecture.
Cela peut produire des comportements inattendus.
Django propose transaction.on_commit() précisément pour déclencher une action après validation de la transaction.
from django.db import transaction
with transaction.atomic():
user = User.objects.create(...)
transaction.on_commit(
lambda: envoyer_email_bienvenue.delay(user.id)
)
Cette petite différence est extrêmement importante.
L’idée devient :
Transaction
|
v
COMMIT
|
v
enqueue Celery
au lieu de :
Transaction
|
+---- enqueue Celery
|
v
COMMIT
Cette notion devient particulièrement importante dans les applications e-commerce, financières et métiers où les tâches dépendent de données fraîchement créées.
Éviter de capturer des variables qui deviennent obsolètes
Avec :
transaction.on_commit(
lambda: envoyer_email_bienvenue.delay(user.id)
)
on capture une valeur simple :
user.id
C’est préférable à une logique complexe dépendant d’un objet dont l’état peut changer.
Une approche lisible :
user_id = user.id
transaction.on_commit(
lambda: envoyer_email_bienvenue.delay(user_id)
)
Cela permet de rendre l’intention plus explicite.
Générer des fichiers en arrière-plan
Un excellent cas d’usage de Celery consiste à générer un rapport.
Supposons un modèle :
class Report(models.Model):
name = models.CharField(max_length=255)
file = models.FileField(upload_to="reports/", null=True, blank=True)
status = models.CharField(
max_length=30,
default="pending",
)
created_at = models.DateTimeField(auto_now_add=True)
La tâche :
from celery import shared_task
from django.core.files.base import ContentFile
from .models import Report
@shared_task
def generer_rapport(report_id):
report = Report.objects.get(pk=report_id)
report.status = "processing"
report.save(update_fields=["status"])
contenu = construire_rapport(report)
report.file.save(
f"report-{report.id}.txt",
ContentFile(contenu.encode("utf-8")),
save=False,
)
report.status = "completed"
report.save(
update_fields=["file", "status"],
)
La vue :
def create_report(request):
report = Report.objects.create(
name="Rapport mensuel"
)
transaction.on_commit(
lambda: generer_rapport.delay(report.id)
)
return JsonResponse({
"id": report.id,
"status": report.status,
})
Le navigateur reçoit immédiatement :
{
"id": 42,
"status": "pending"
}
Puis le worker effectue le traitement.
Gérer correctement l’état métier
Une erreur fréquente consiste à dépendre uniquement de l’état Celery :
PENDING
STARTED
SUCCESS
FAILURE
Pour de nombreuses applications métier, il est préférable d’avoir aussi un état propre au domaine :
pending
processing
completed
failed
Pourquoi ?
Parce que Celery sait qu’une tâche est terminée, mais votre application doit savoir ce que cela signifie métier.
Par exemple :
class ImportJob(models.Model):
status = models.CharField(
max_length=30,
choices=[
("pending", "Pending"),
("processing", "Processing"),
("completed", "Completed"),
("failed", "Failed"),
],
default="pending",
)
Le worker :
@shared_task(bind=True)
def importer_fichier(self, import_job_id):
job = ImportJob.objects.get(pk=import_job_id)
job.status = "processing"
job.save(update_fields=["status"])
try:
traiter_import(job)
except Exception:
job.status = "failed"
job.save(update_fields=["status"])
raise
job.status = "completed"
job.save(update_fields=["status"])
Cette architecture permet au front-end de consulter votre propre modèle métier plutôt que de dépendre complètement du backend de résultats Celery.
Pourquoi stocker les résultats Celery peut être une mauvaise idée
Le result backend est utile, mais toutes les tâches n’ont pas besoin de conserver leur valeur de retour.
Imaginez :
@shared_task
def envoyer_newsletter(user_id):
send_email(...)
return {
"success": True,
"user_id": user_id,
"full_email_content": "...",
}
Si vous envoyez des centaines de milliers de tâches et stockez de gros résultats, vous pouvez remplir Redis inutilement.
Une tâche comme :
@shared_task
def envoyer_email(user_id):
...
n’a parfois pas besoin d’un résultat détaillé.
Selon votre architecture, vous pouvez décider de conserver seulement les informations métier nécessaires dans la base Django.
La documentation Celery rappelle d’ailleurs que Redis est très efficace comme backend de résultats, mais qu’il faut tenir compte de la mémoire disponible et de la gestion de la persistance.
Configurer une expiration des résultats
Les résultats de tâches ne devraient pas nécessairement être conservés éternellement.
Vous pouvez configurer :
CELERY_RESULT_EXPIRES = 3600
Cela indique une durée de conservation des résultats.
Pour une application qui utilise surtout ses propres modèles Django pour suivre les traitements, une durée relativement courte peut être suffisante.
Cela évite de transformer Redis en gigantesque base historique de toutes les tâches exécutées depuis plusieurs années.
Les tâches périodiques avec Celery Beat
Certaines opérations ne sont pas déclenchées par un utilisateur.
Par exemple :
tous les jours à minuit
toutes les 5 minutes
toutes les heures
chaque lundi
Celery fournit Celery Beat pour planifier ce type de tâche.
Vous pouvez définir un scheduler dans la configuration.
Par exemple :
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
"nettoyer-les-sessions": {
"task": "comptes.tasks.nettoyer_sessions",
"schedule": crontab(hour=0, minute=0),
},
}
La tâche :
from celery import shared_task
@shared_task
def nettoyer_sessions():
...
Puis lancer Beat :
celery -A monprojet beat --loglevel=info
Et le worker :
celery -A monprojet worker --loglevel=info
Beat ne remplace pas le worker. Beat programme les messages. Le worker exécute les tâches.
Cela donne :
Celery Beat
|
v
Redis
|
v
Celery Worker
Installer django-celery-beat
Pour les projets qui souhaitent gérer les tâches périodiques depuis l’administration Django, django-celery-beat est une extension fréquemment utilisée.
Après installation :
pip install django-celery-beat
Ajoutez :
INSTALLED_APPS = [
...
"django_celery_beat",
]
Puis :
python manage.py migrate
Vous pouvez ensuite démarrer Beat avec un scheduler basé sur la base de données :
celery -A monprojet beat \
--loglevel=info \
--scheduler django_celery_beat.schedulers:DatabaseScheduler
Cette approche peut être particulièrement pratique lorsqu’une équipe souhaite modifier les horaires depuis le panneau d’administration sans redéployer l’application.
Planifier une tâche quotidienne
Supposons :
@shared_task
def synchroniser_donnees():
...
Une entrée périodique pourrait exécuter cette tâche chaque jour.
Le principe est :
00:00
|
v
Beat
|
v
Redis
|
v
Worker
|
v
synchroniser_donnees()
Il est toujours préférable de rendre la tâche idempotente.
Pourquoi ?
Parce que si elle est déclenchée deux fois, l’application doit rester dans un état cohérent.
Configurer le fuseau horaire
Dans Django :
TIME_ZONE = "Africa/Casablanca"
USE_TZ = True
Et pour Celery :
CELERY_TIMEZONE = "Africa/Casablanca"
CELERY_ENABLE_UTC = True
Le temps est une source classique de bugs dans les tâches planifiées.
Une tâche programmée pour :
00:00
doit être interprétée dans le fuseau attendu.
Les changements d’heure, les conversions UTC et les environnements de développement différents peuvent devenir particulièrement difficiles à diagnostiquer si les paramètres sont incohérents.
Les signaux Django et Celery
Vous pouvez également déclencher des tâches à partir d’événements Django.
Par exemple :
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import User
from .tasks import envoyer_email_bienvenue
@receiver(post_save, sender=User)
def user_created(sender, instance, created, **kwargs):
if created:
envoyer_email_bienvenue.delay(instance.id)
Cela fonctionne, mais il faut être prudent avec les signaux.
Les signaux peuvent rendre le flux métier difficile à suivre :
save()
|
+-- signal
|
+-- Celery
|
+-- Email
Dans des applications complexes, une logique explicite est souvent plus facile à maintenir :
user = create_user(...)
transaction.on_commit(
lambda: envoyer_email_bienvenue.delay(user.id)
)
Le signal peut rester utile dans certains domaines, mais il ne faut pas automatiquement cacher une dépendance métier essentielle derrière un mécanisme implicite.
Tâches composées avec chain
Celery permet de composer plusieurs tâches.
Exemple :
from celery import chain
workflow = chain(
telecharger_document.s(document_id),
convertir_document.s(),
analyser_document.s(),
)
workflow.apply_async()
Le principe devient :
Télécharger
|
v
Convertir
|
v
Analyser
La signature .s() permet de préparer une tâche pour la composition.
Cela peut être utile lorsque chaque étape produit un résultat consommé par la suivante.
Utiliser group
Un group permet de lancer plusieurs tâches en parallèle.
from celery import group
workflow = group(
traiter_client.s(1),
traiter_client.s(2),
traiter_client.s(3),
traiter_client.s(4),
)
result = workflow.apply_async()
On obtient :
+--> Client 1
|
Group -------+--> Client 2
|
+--> Client 3
|
+--> Client 4
C’est particulièrement pratique lorsque plusieurs éléments peuvent être traités indépendamment.
Utiliser chord pour attendre plusieurs résultats
Une chord permet d’exécuter plusieurs tâches puis une tâche finale.
from celery import chord
workflow = chord(
[
analyser.s(1),
analyser.s(2),
analyser.s(3),
]
)(aggreger_resultats.s())
Le schéma devient :
analyser 1
|
analyser 2
|
analyser 3
|
v
aggreger_resultats
Ce modèle est très pratique pour des pipelines de traitement de données.
Le piège des tâches trop volumineuses
Il est tentant de faire :
@shared_task
def traiter_tout():
for item in BigTable.objects.all():
traiter(item)
Cela paraît simple, mais peut devenir un cauchemar.
Si vous avez :
2 000 000 lignes
une seule tâche peut durer des heures.
Une meilleure stratégie consiste souvent à découper :
@shared_task
def traiter_lot(ids):
for item_id in ids:
traiter_item(item_id)
Puis :
for batch in batches:
traiter_lot.delay(batch)
Vous obtenez alors :
2 000 000 éléments
|
+-- lot 1
+-- lot 2
+-- lot 3
+-- lot 4
...
Les petites tâches sont généralement plus faciles à retry, à superviser et à distribuer.
Faire attention à la taille des messages Redis
Le broker transporte des messages.
Il est donc déconseillé de faire :
traiter_document.delay(mega_octets_de_donnees)
Préférez :
traiter_document.delay(document_id)
puis rechargez les données côté worker.
La documentation Celery recommande de considérer Redis surtout pour des messages de petite taille et rapides à transporter ; des messages volumineux peuvent créer de la congestion.
C’est un principe simple qui évite beaucoup de problèmes :
Message Celery -> référence légère
Stockage réel -> base / objet / fichier
Sécurité : ne transmettez pas de secrets dans les arguments
Évitez :
envoyer_api.delay(
api_key="super-secret",
)
Les arguments de tâche deviennent des données transportées par l’infrastructure de messages.
Il vaut mieux stocker les secrets dans :
variables d’environnement
secret manager
configuration sécurisée
et récupérer le secret dans le worker.
Par exemple :
import os
API_KEY = os.environ["API_KEY"]
Puis :
@shared_task
def appeler_service(client_id):
response = requests.get(
"https://api.example.com",
headers={
"Authorization": f"Bearer {API_KEY}",
},
)
Sécuriser Redis
Une erreur fréquente consiste à exposer Redis directement sur Internet.
Redis ne doit pas devenir :
Internet
|
v
Redis ouvert
Préférez :
Internet
|
v
Django
|
+-- réseau privé --> Redis
|
+-- réseau privé --> Workers
Selon votre environnement, utilisez un réseau privé, des règles de firewall, une authentification adaptée et TLS lorsque cela est nécessaire.
Lorsque Redis est utilisé sur TLS, Celery documente l’utilisation du protocole rediss:// pour le backend Redis avec les paramètres SSL associés.
Exemple :
CELERY_BROKER_URL = (
"rediss://username:password@redis.example.com:6380/0"
)
La configuration exacte dépend évidemment du service Redis utilisé.
Utiliser des variables d’environnement
Ne mettez pas :
CELERY_BROKER_URL = "redis://user:password@server:6379/0"
directement dans Git.
Préférez :
import os
CELERY_BROKER_URL = os.environ["CELERY_BROKER_URL"]
CELERY_RESULT_BACKEND = os.environ["CELERY_RESULT_BACKEND"]
Dans .env :
CELERY_BROKER_URL=redis://127.0.0.1:6379/0
CELERY_RESULT_BACKEND=redis://127.0.0.1:6379/1
Puis utilisez un mécanisme adapté pour charger ces variables.
Configurer un projet avec Docker Compose
Une architecture de développement courante :
services:
web:
build: .
command: python manage.py runserver 0.0.0.0:8000
volumes:
- .:/app
ports:
- "8000:8000"
depends_on:
- redis
- db
worker:
build: .
command: celery -A monprojet worker --loglevel=info
volumes:
- .:/app
depends_on:
- redis
- db
beat:
build: .
command: celery -A monprojet beat --loglevel=info
volumes:
- .:/app
depends_on:
- redis
- db
redis:
image: redis:7
db:
image: postgres:17
environment:
POSTGRES_DB: app
POSTGRES_USER: app
POSTGRES_PASSWORD: app
Cela donne :
+----------------+
| web |
+-------+--------+
|
v
+---------+
| Redis |
+----+----+
|
+-------+-------+
| |
v v
+---------+ +---------+
| worker | | beat |
+---------+ +---------+
+---------+
| postgres|
+---------+
Cette configuration est excellente pour comprendre l’architecture.
En production, vous pourrez remplacer certains services par des systèmes managés ou des conteneurs orchestrés selon vos besoins.
Le Dockerfile
Un Dockerfile simple :
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "manage.py", "runserver", "0.0.0.0:8000"]
Les workers utiliseront la même image, mais avec une commande différente :
command: celery -A monprojet worker --loglevel=info
Cette approche évite de dupliquer l’environnement Python.
Une architecture de production avec Gunicorn et Celery
Une architecture classique pourrait ressembler à :
Internet
|
v
Nginx / LB
|
+-------+-------+
| |
v v
Django 1 Django 2
| |
+-------+-------+
|
v
Redis
|
+------------+------------+
| | |
v v v
Worker 1 Worker 2 Worker 3
|
v
Database
Le serveur Django ne doit pas exécuter lui-même les tâches longues.
Le worker doit être un processus indépendant.
C’est important pour plusieurs raisons : redémarrage indépendant, scaling séparé, limitations de ressources distinctes et isolation.
Gunicorn ne remplace pas Celery
Gunicorn gère les requêtes HTTP.
Celery gère les tâches.
Un serveur Gunicorn avec 4 workers :
gunicorn monprojet.wsgi:application --workers 4
ne signifie pas que vous disposez de quatre workers Celery.
Il faut démarrer Celery séparément :
celery -A monprojet worker --loglevel=info
On peut donc avoir :
Gunicorn
|
+-- 4 processus web
Celery
|
+-- 4 processus task workers
Ce sont des responsabilités différentes.
Supervisor pour les workers
Sur un serveur classique, Supervisor peut maintenir les processus actifs.
Exemple :
[program:celery]
command=/var/www/app/.venv/bin/celery -A monprojet worker --loglevel=info
directory=/var/www/app
user=www-data
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/celery.log
Pour Beat :
[program:celery-beat]
command=/var/www/app/.venv/bin/celery -A monprojet beat --loglevel=info
directory=/var/www/app
user=www-data
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/celery-beat.log
Dans un environnement utilisant systemd, on peut également créer des unités dédiées.
Exemple avec systemd
[Unit]
Description=Celery Worker
After=network.target redis.service
[Service]
Type=simple
User=www-data
WorkingDirectory=/var/www/app
Environment="PATH=/var/www/app/.venv/bin"
ExecStart=/var/www/app/.venv/bin/celery \
-A monprojet worker \
--loglevel=INFO
Restart=always
[Install]
WantedBy=multi-user.target
Puis :
sudo systemctl daemon-reload
sudo systemctl enable celery
sudo systemctl start celery
Vérification :
sudo systemctl status celery
Le détail des commandes dépend évidemment de votre distribution et de votre méthode de déploiement, mais le principe reste identique : Celery devient un service indépendant.
Monitoring avec Flower
Pour observer les workers, les tâches et leurs états, Flower est un outil populaire dans l’écosystème Celery.
Installation :
pip install flower
Puis :
celery -A monprojet flower
Vous pourrez généralement accéder à une interface Web permettant de visualiser les workers, les tâches et certaines métriques.
Dans un environnement de développement, cela est très agréable.
En production, il faut néanmoins sécuriser l’accès à cette interface. Une interface de monitoring exposée sans protection peut révéler des informations sensibles sur votre infrastructure.
Ce qu’il faut surveiller en production
Une architecture Celery saine ne se résume pas à voir « worker online ».
Vous devriez idéalement surveiller :
- nombre de tâches en attente
- temps moyen d’exécution
- nombre d’échecs
- taux de retry
- tâches très anciennes
- consommation CPU
- consommation RAM
- taille des files
- mémoire Redis
- latence Redis
- disponibilité des workers
Une file qui grossit constamment est un signal important.
Supposons :
09:00 -> 100 tâches
10:00 -> 800 tâches
11:00 -> 4 000 tâches
12:00 -> 18 000 tâches
Même si Django répond parfaitement aux requêtes HTTP, votre système asynchrone est en train d’accumuler une dette de traitement.
Le phénomène du backlog
Un backlog signifie simplement qu’il y a plus de travail entrant que de travail traité.
On peut le représenter :
Production des tâches
100 / minute
Traitement
80 / minute
Différence
20 / minute
La file va grossir inexorablement.
Il faut alors :
- réduire le coût des tâches
- augmenter le nombre de workers
- optimiser la base
- supprimer du travail inutile
- utiliser plusieurs queues
- améliorer les appels externes
Ajouter des workers peut aider, mais seulement si la ressource limitante le permet.
Le piège de la concurrence excessive
Supposons que vous avez :
CPU : 4 cœurs
et que vous démarrez :
celery -A monprojet worker --concurrency=32
Cela ne signifie pas forcément que votre application sera 8 fois plus rapide.
Vous pouvez au contraire obtenir :
CPU saturation
context switching
memory pressure
database saturation
external API throttling
Le dimensionnement doit donc être basé sur des mesures réelles.
Attention à la base de données
Celery peut augmenter la vitesse de traitement, mais si toutes vos tâches effectuent :
Model.objects.filter(...)
en masse, la base devient rapidement le goulot d’étranglement.
Imaginez 20 workers exécutant simultanément :
for item in items:
Model.objects.get(...)
Vous avez peut-être accidentellement créé un problème de type N+1 multiplié par 20.
Optimisez donc les tâches exactement comme vous optimiseriez les vues Django :
select_related()
prefetch_related()
bulk_create()
bulk_update()
update()
et les index de base de données.
Exemple de tâche optimisée
Mauvaise version :
@shared_task
def traiter_commandes():
for order in Order.objects.all():
user = order.user
total = calculer_total(order)
enregistrer_total(order, total)
Vous pouvez parfois améliorer :
@shared_task
def traiter_commandes():
orders = (
Order.objects
.select_related("user")
.iterator(chunk_size=500)
)
for order in orders:
total = calculer_total(order)
enregistrer_total(order, total)
Le détail dépend du modèle et des opérations exécutées, mais l’idée est de réfléchir à la quantité de données chargées en mémoire et au nombre de requêtes.
Éviter les tâches qui changent trop de choses sans contrôle
Une énorme tâche transactionnelle :
@shared_task
def migration_massive():
with transaction.atomic():
for row in rows:
...
peut garder une transaction ouverte pendant très longtemps.
Une transaction longue peut provoquer :
locks
bloquages
bloat
timeouts
Découper le traitement en petits lots est souvent préférable.
Par exemple :
@shared_task
def traiter_lot(ids):
with transaction.atomic():
...
puis plusieurs tâches indépendantes.
La gestion des doublons
Supposons qu’un bouton dans une interface soit cliqué deux fois.
Votre vue peut créer deux tâches :
envoyer_facture.delay(invoice.id)
envoyer_facture.delay(invoice.id)
Même si l’utilisateur n’a voulu qu’une seule opération.
L’idempotence protège partiellement contre cette situation, mais certaines tâches nécessitent une logique plus explicite.
Vous pouvez utiliser un modèle de suivi :
class TaskExecution(models.Model):
key = models.CharField(
max_length=255,
unique=True,
)
created_at = models.DateTimeField(
auto_now_add=True,
)
Puis utiliser une clé métier :
invoice:42:send
L’application peut vérifier si cette opération existe déjà.
Créer une clé d’idempotence
Exemple :
@shared_task
def envoyer_facture(invoice_id):
key = f"invoice:{invoice_id}:send"
created = TaskExecution.objects.get_or_create(
key=key
)
if not created:
return "already processed"
...
Il faut cependant faire attention à la manière dont cette logique est écrite, notamment aux conditions de concurrence.
Dans les systèmes sensibles, une contrainte unique en base de données est beaucoup plus fiable qu’un simple :
if not exists():
create()
car deux workers peuvent exécuter ce code simultanément.
L’importance des contraintes de base de données
Celery ne remplace pas les garanties de la base.
Si une donnée doit être unique :
class Payment(models.Model):
transaction_id = models.CharField(
max_length=255,
unique=True,
)
La base garantit réellement l’unicité.
Un simple test Python n’offre pas la même garantie sous concurrence.
Cela devient particulièrement important lorsqu’un système Celery traite plusieurs tâches simultanément.
Gestion des tâches perdues et redélivrées
Les systèmes distribués doivent gérer les pannes.
Imaginez :
Worker reçoit tâche
|
v
Commence traitement
|
X
Worker crash
Selon la configuration du broker, de l’acknowledgement et du transport, une tâche peut être redélivrée.
C’est pourquoi il faut éviter les hypothèses comme :
« Une tâche est forcément exécutée une seule fois. »
Dans un environnement distribué, il est beaucoup plus sain de concevoir les tâches pour tolérer les répétitions lorsque cela est possible.
Visibility timeout avec Redis
Le transport Redis de Celery possède une notion de visibility_timeout : le message peut être redélivré si le worker ne l’a pas acquitté dans la fenêtre prévue. La documentation Celery décrit cette configuration et précise que sa valeur doit être prise en compte notamment pour les tâches dont l’exécution peut être longue.
Cela signifie qu’une tâche extrêmement longue doit être pensée avec attention.
Exemple :
CELERY_BROKER_TRANSPORT_OPTIONS = {
"visibility_timeout": 3600,
}
Si vos tâches peuvent durer plusieurs heures, une valeur adaptée peut être nécessaire.
Mais augmenter aveuglément le timeout n’est pas une solution magique. Il faut comprendre le comportement global des retries, des acknowledgements et du type de transport utilisé.
Les tâches longues doivent être conçues différemment
Une tâche de cinq minutes n’est pas forcément inquiétante.
Une tâche de six heures mérite beaucoup plus de réflexion.
Vous devrez probablement vous demander :
Puis-je la découper ?
Puis-je reprendre après interruption ?
Puis-je enregistrer la progression ?
Puis-je faire un checkpoint ?
Puis-je paralléliser ?
Puis-je éviter de refaire le travail déjà accompli ?
Au lieu de :
@shared_task
def traiter_tout():
traiter_million_d_elements()
vous pouvez avoir :
@shared_task
def traiter_lot(offset, limit):
...
et plusieurs tâches :
traiter_lot.delay(0, 1000)
traiter_lot.delay(1000, 1000)
traiter_lot.delay(2000, 1000)
Cette stratégie facilite énormément la reprise.
Enregistrer la progression
Pour une interface utilisateur, vous pouvez stocker :
class ImportJob(models.Model):
total = models.PositiveIntegerField(default=0)
processed = models.PositiveIntegerField(default=0)
status = models.CharField(max_length=30)
La tâche :
@shared_task
def import_job(job_id):
job = ImportJob.objects.get(pk=job_id)
items = charger_items(job)
job.total = len(items)
job.status = "processing"
job.save(update_fields=["total", "status"])
for index, item in enumerate(items, start=1):
traiter(item)
job.processed = index
if index % 100 == 0:
job.save(update_fields=["processed"])
job.status = "completed"
job.save(update_fields=["status"])
Le front-end peut alors afficher :
37 %
ou :
3700 / 10000
C’est souvent plus utile qu’un simple :
STARTED
Ne pas mettre une boucle infinie dans Celery
Une tâche comme :
@shared_task
def boucle():
while True:
...
est presque toujours un mauvais signal.
Un worker est une ressource de traitement.
Les opérations permanentes doivent être gérées par un service conçu pour cela, pas par une tâche Celery qui monopolise un worker indéfiniment.
Celery est excellent pour :
job -> execute -> done
et beaucoup moins adapté à :
processus permanent -> tourner pour toujours
Exemple réel : traitement d’images
Prenons un modèle :
class Photo(models.Model):
original = models.ImageField(upload_to="photos/")
thumbnail = models.ImageField(
upload_to="photos/thumbs/",
null=True,
blank=True,
)
La tâche :
from io import BytesIO
from celery import shared_task
from django.core.files.base import ContentFile
from PIL import Image
from .models import Photo
@shared_task
def generer_thumbnail(photo_id):
photo = Photo.objects.get(pk=photo_id)
with photo.original.open("rb") as file:
image = Image.open(file)
image.thumbnail((400, 400))
output = BytesIO()
image.save(output, format="JPEG", quality=85)
photo.thumbnail.save(
f"thumb-{photo.id}.jpg",
ContentFile(output.getvalue()),
save=False,
)
photo.save(update_fields=["thumbnail"])
La vue peut immédiatement retourner :
photo = Photo.objects.create(
original=request.FILES["image"]
)
transaction.on_commit(
lambda: generer_thumbnail.delay(photo.id)
)
L’utilisateur n’a pas besoin d’attendre la création de toutes les variantes.
Ne faites pas transiter le contenu binaire via Redis
Évitez :
generer_thumbnail.delay(image_bytes)
Préférez :
generer_thumbnail.delay(photo.id)
Le worker récupère ensuite le fichier depuis le stockage.
Dans les applications Django modernes, le stockage peut être :
filesystem
S3
MinIO
Azure Blob
Google Cloud Storage
Celery n’a pas besoin de transporter les octets du fichier eux-mêmes.
Exemple : scraper une API externe
Supposons :
@shared_task(
bind=True,
autoretry_for=(TimeoutError,),
retry_backoff=True,
retry_kwargs={"max_retries": 4},
)
def synchroniser_produit(self, product_id):
product = Product.objects.get(pk=product_id)
response = requests.get(
product.external_url,
timeout=10,
)
response.raise_for_status()
data = response.json()
product.price = data["price"]
product.available = data["available"]
product.save(
update_fields=[
"price",
"available",
]
)
Remarquez plusieurs bonnes pratiques :
timeout explicite
retry limité
identifiant léger
mise à jour ciblée
Sans timeout, une requête externe peut rester bloquée beaucoup plus longtemps que prévu.
Toujours mettre des timeouts sur les appels externes
Évitez :
requests.get(url)
Préférez :
requests.get(
url,
timeout=10,
)
Avec une tâche Celery, l’absence de timeout est encore plus problématique parce que vous pouvez accumuler des workers bloqués.
Un bon système asynchrone doit également savoir dire :
cette dépendance externe est trop lente, j’abandonne cette tentative et je réessaierai plus tard.
Les erreurs d’API externes
Vous devriez distinguer :
400 -> mauvais appel
401 -> authentification
403 -> autorisation
404 -> ressource inexistante
429 -> rate limit
500 -> erreur serveur
502/503/504 -> problème temporaire
Toutes ces erreurs ne doivent pas être traitées de la même façon.
Pour un :
404
réessayer dix fois est probablement inutile.
Pour :
503
un retry avec backoff peut être logique.
Pour :
429
vous devez respecter la limitation du fournisseur et idéalement le mécanisme Retry-After lorsque disponible.
Contrôler les rate limits
Imaginons que votre fournisseur autorise :
100 requêtes / minute
Si vous démarrez dix workers qui lancent chacun 50 requêtes par minute, vous dépassez largement la limite.
Celery permet certaines stratégies de limitation de débit.
Vous pouvez par exemple configurer un rate limit sur une tâche :
@shared_task(rate_limit="10/m")
def appeler_service_externe():
...
Les rate limits doivent cependant être considérés dans le contexte du nombre de workers, de leurs files et du fournisseur externe.
La question n’est pas simplement :
rate_limit = 10/m
mais :
10/m par worker ou dans quelle architecture ?
Il faut donc penser à l’échelle globale.
Logging dans les tâches Celery
Évitez les dizaines de :
print(...)
Préférez le logging Python :
import logging
logger = logging.getLogger(__name__)
@shared_task
def synchroniser_produit(product_id):
logger.info(
"Synchronisation du produit %s",
product_id,
)
Pour une erreur :
try:
...
except Exception:
logger.exception(
"Échec de synchronisation du produit %s",
product_id,
)
raise
logger.exception() inclut la trace d’exception lorsque l’appel est effectué dans un bloc except.
Cela facilite énormément le diagnostic.
Ajouter un identifiant métier dans les logs
Au lieu d’un log générique :
Task failed
préférez :
logger.error(
"Impossible de générer le rapport %s",
report_id,
)
Encore mieux, si vous avez :
user_id
order_id
task_id
job_id
vous pouvez corréler les événements.
Cela devient particulièrement utile lorsque plusieurs workers exécutent des centaines de tâches simultanément.
Attraper ou laisser remonter les exceptions ?
Une erreur fréquente :
@shared_task
def ma_tache():
try:
...
except Exception as exc:
print(exc)
Puis la fonction ne relève jamais l’exception.
Le résultat est trompeur : Celery peut considérer la tâche comme réussie alors qu’une opération essentielle a échoué.
Préférez :
@shared_task
def ma_tache():
try:
...
except Exception:
logger.exception("Erreur dans ma_tache")
raise
Ainsi, Celery sait que la tâche a réellement échoué.
Utiliser des tâches robustes avec self.request
Avec :
@shared_task(bind=True)
def ma_tache(self):
print(self.request.id)
vous pouvez récupérer des informations sur la tâche courante.
Par exemple :
@shared_task(bind=True)
def traiter(self, item_id):
logger.info(
"Task %s traite item %s",
self.request.id,
item_id,
)
L’identifiant de tâche devient très utile pour diagnostiquer un problème particulier.
Concevoir une interface métier plutôt qu’une API Celery exposée
Il est rarement souhaitable d’exposer directement les détails internes de Celery à tous les clients.
Plutôt que :
GET /celery-result/UUID
vous pouvez avoir :
GET /api/reports/42/
qui renvoie :
{
"id": 42,
"status": "processing",
"progress": 67,
"download_url": null
}
Puis, une fois terminé :
{
"id": 42,
"status": "completed",
"progress": 100,
"download_url": "/media/reports/42.pdf"
}
Le client connaît votre domaine, pas nécessairement les détails techniques de Celery.
Celery et les API REST Django
Une API Django REST Framework peut lancer une tâche :
from rest_framework.response import Response
from rest_framework.views import APIView
from .models import Report
from .tasks import generer_rapport
class ReportGenerateView(APIView):
def post(self, request):
report = Report.objects.create(
name=request.data["name"]
)
transaction.on_commit(
lambda: generer_rapport.delay(report.id)
)
return Response(
{
"id": report.id,
"status": "pending",
},
status=202,
)
Le code HTTP 202 Accepted est particulièrement adapté à ce concept : la demande a été acceptée mais n’est pas encore terminée.
C’est un modèle très propre pour les opérations longues.
Connecter un front-end Next.js
Imaginez votre front-end :
const response = await fetch("/api/reports/", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({
name: "Rapport annuel",
}),
});
const report = await response.json();
Vous récupérez :
{
"id": 42,
"status": "pending"
}
Puis :
async function pollReport(id) {
while (true) {
const response = await fetch(`/api/reports/${id}/`);
const data = await response.json();
if (data.status === "completed") {
return data;
}
if (data.status === "failed") {
throw new Error("La génération a échoué");
}
await new Promise(resolve => setTimeout(resolve, 2000));
}
}
Dans une vraie application, il faudra naturellement gérer l’annulation, les timeouts du polling, la fermeture de page et éventuellement préférer les WebSockets ou SSE.
Tester les tâches Celery
Tester une tâche :
from django.test import TestCase
from .tasks import additionner
class AdditionnerTaskTests(TestCase):
def test_addition(self):
result = additionner.run(2, 3)
self.assertEqual(
result,
5,
)
Vous pouvez appeler directement :
additionner.run(...)
pour tester la fonction sans avoir besoin de Redis ou d’un worker pour chaque test unitaire.
Dans d’autres cas, vous pouvez tester l’intégration complète avec un environnement Celery dédié.
L’essentiel est de distinguer :
tests unitaires
tests d’intégration
tests end-to-end
Tester les retries
Une tâche avec :
@shared_task(bind=True, max_retries=3)
def ma_tache(self):
...
doit être testée en simulant les erreurs temporaires.
Le but n’est pas simplement de vérifier « la tâche fonctionne ».
Il faut aussi tester :
que se passe-t-il au premier échec ?
que se passe-t-il au deuxième ?
quand la tâche arrête-t-elle de retry ?
que se passe-t-il après le dernier échec ?
Tester l’idempotence
Une excellente stratégie de test est d’appeler la même tâche deux fois.
traiter_commande.run(order_id)
traiter_commande.run(order_id)
Puis vérifier que l’état métier ne change pas de façon incorrecte.
Exemple :
self.assertEqual(
payment.status,
"paid",
)
et surtout :
self.assertEqual(
Payment.objects.filter(
transaction_id=transaction_id
).count(),
1,
)
Cela permet de vérifier que les contraintes métier résistent à la répétition.
Les migrations et les workers
Lors d’un déploiement, il faut être conscient qu’un worker Celery peut continuer à exécuter du code ancien pendant qu’une migration de base change le schéma.
Supposons :
Version A
Worker A
Database A
puis vous déployez :
Version B
Database B
Une tâche créée par Version A peut être exécutée par un worker Version B ou inversement selon le processus de déploiement.
Cela devient important lorsque vous supprimez un champ, renommez une colonne ou changez le format d’une donnée.
Une stratégie de déploiement progressive doit donc préserver temporairement la compatibilité entre versions lorsque nécessaire.
Exemple de déploiement prudent
Au lieu de supprimer immédiatement :
old_field
vous pouvez faire :
Étape 1 : ajouter new_field
Étape 2 : déployer code qui écrit old + new
Étape 3 : migrer les données
Étape 4 : déployer code qui lit new
Étape 5 : arrêter l’usage de old
Étape 6 : supprimer old
Ce genre de stratégie est particulièrement important avec les tâches asynchrones, car leur exécution peut être retardée.
Redis et la persistance
Lorsque Redis est utilisé comme broker et/ou backend, vous devez comprendre le rôle de la persistance.
Le broker contient des messages dont vous pouvez avoir besoin de conserver selon votre architecture.
Le result backend contient des résultats qui peuvent parfois être recréés.
Il ne faut donc pas considérer toutes les données Redis comme ayant exactement la même valeur métier.
Pour des résultats temporaires, Redis peut être excellent.
Pour des données critiques qui doivent survivre durablement à une panne complète, une base de données durable peut être plus adaptée.
La documentation Celery recommande également de considérer d’autres backends lorsque la persistance à long terme des résultats est importante.
Séparer broker et backend
Une configuration simple :
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
Une architecture plus avancée pourrait utiliser :
Broker -> Redis
Result backend -> PostgreSQL
L’avantage est de ne pas accumuler tous les résultats dans la même mémoire Redis.
La documentation Celery mentionne notamment les backends basés sur SQLAlchemy pour des bases SQL telles que MySQL et PostgreSQL, parmi différentes options disponibles.
Choisir Redis ou RabbitMQ
Redis est souvent choisi pour sa simplicité.
RabbitMQ peut être préférable dans certains scénarios de messaging plus complexes.
Celery classe actuellement Redis et RabbitMQ parmi les transports stables, tout en leur attribuant des caractéristiques différentes. La documentation indique notamment que Redis fonctionne très bien pour des messages rapides et petits, tandis que RabbitMQ gère mieux les messages plus volumineux.
Le choix doit être guidé par :
volume
taille des messages
fonctionnalités de routage
haute disponibilité
opérations
compétences de l’équipe
infrastructure existante
Redis est souvent un excellent choix pour commencer.
Pourquoi Redis est très populaire avec Django et Celery
Il y a quelque chose de profondément pratique dans cette combinaison.
Django connaît déjà très bien le monde Python.
Celery s’intègre naturellement à Django.
Redis est relativement simple à lancer.
La configuration minimale tient dans quelques lignes :
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
Et le worker démarre avec :
celery -A monprojet worker -l info
Cette simplicité explique pourquoi la combinaison est si populaire pour des projets de taille petite à moyenne et également pour de nombreux systèmes plus importants.
Une configuration de settings plus propre
Voici un exemple plus complet :
import os
CELERY_BROKER_URL = os.getenv(
"CELERY_BROKER_URL",
"redis://127.0.0.1:6379/0",
)
CELERY_RESULT_BACKEND = os.getenv(
"CELERY_RESULT_BACKEND",
"redis://127.0.0.1:6379/1",
)
CELERY_ACCEPT_CONTENT = [
"json",
]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "Africa/Casablanca"
CELERY_TASK_TRACK_STARTED = True
CELERY_RESULT_EXPIRES = 3600
CELERY_TASK_TIME_LIMIT = 1800
CELERY_TASK_SOFT_TIME_LIMIT = 1500
CELERY_BROKER_TRANSPORT_OPTIONS = {
"visibility_timeout": 3600,
}
Cette configuration est un point de départ, pas une vérité universelle. Les valeurs de timeouts et de visibilité doivent correspondre à votre charge et à la durée réelle de vos tâches.
Pourquoi JSON est un choix raisonnable
Celery utilise par défaut JSON comme type de contenu accepté dans sa configuration moderne, et sa documentation recommande de contrôler explicitement les sérialiseurs acceptés.
Vous pouvez définir :
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
Cela force les messages à respecter une forme de sérialisation explicite.
Évitez d’activer des sérialiseurs dangereux sans raison.
La documentation Celery rappelle notamment que l’utilisation de formats comme pickle ou YAML avec des sources non fiables demande une attention particulière à la sécurité.
Ne jamais faire confiance aux données d’une tâche
Une tâche Celery n’est pas un sanctuaire.
Si une donnée peut provenir d’une entrée utilisateur :
@shared_task
def executer_commande(command):
...
ne supposez pas que command est sûr parce qu’il vient de Redis.
Validez toujours :
user_id
object_id
path
URL
paramètres externes
comme vous le feriez dans une requête HTTP.
Gérer les chemins de fichiers avec prudence
Évitez de permettre à une tâche de recevoir un chemin arbitraire :
traiter_fichier.delay("../../etc/passwd")
Le worker doit contrôler les chemins autorisés.
Une approche préférable est :
traiter_fichier.delay(file_id)
puis récupérer le chemin depuis un modèle contrôlé.
Les jobs et les tâches : deux concepts différents
Dans les petits projets, vous pouvez vous contenter de Celery.
Dans une application métier plus importante, vous aurez souvent intérêt à créer un modèle :
class Job(models.Model):
task_id = models.CharField(
max_length=255,
unique=True,
)
kind = models.CharField(max_length=100)
status = models.CharField(max_length=30)
progress = models.PositiveIntegerField(default=0)
created_at = models.DateTimeField(
auto_now_add=True
)
updated_at = models.DateTimeField(
auto_now=True
)
Lorsque vous enfilez la tâche :
result = generer_rapport.delay(report.id)
Job.objects.create(
task_id=result.id,
kind="report",
status="pending",
)
Cette abstraction facilite ensuite :
historique
permissions
audit
progression
annulation
affichage utilisateur
recherche
statistiques
Celery devient le moteur technique tandis que Job devient le modèle métier.
Ajouter l’utilisateur propriétaire d’un job
Vous pouvez aller plus loin :
class Job(models.Model):
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
)
task_id = models.CharField(
max_length=255,
unique=True,
)
status = models.CharField(
max_length=30,
)
Ainsi, votre API peut vérifier :
job.user == request.user
avant d’afficher le statut.
Cela évite qu’un utilisateur puisse interroger les jobs d’un autre en devinant un identifiant.
Annuler une tâche Celery
Celery permet certaines opérations de contrôle, mais l’annulation réelle d’une tâche doit être abordée avec prudence.
Une tâche déjà en cours d’exécution ne s’arrête pas nécessairement de manière propre simplement parce que vous avez demandé son annulation.
Pour les traitements métiers importants, une approche coopérative est souvent meilleure.
Vous pouvez stocker :
cancel_requested = models.BooleanField(default=False)
Puis dans votre tâche :
for item in items:
job.refresh_from_db(fields=["cancel_requested"])
if job.cancel_requested:
job.status = "cancelled"
job.save(update_fields=["status"])
return
traiter(item)
La tâche devient coopérative.
Elle vérifie périodiquement si elle doit s’arrêter.
Les erreurs partielles
Imaginons un import de 10 000 produits.
Le produit 1 à 5000 fonctionne.
Le produit 5001 échoue.
Que voulez-vous ?
A. Tout annuler
B. Continuer les autres
C. Retry seulement le 5001
D. Marquer le job comme partiellement réussi
Il n’existe pas une seule bonne réponse.
Tout dépend du domaine.
Une tâche de synchronisation peut souvent supporter un traitement partiel.
Une tâche financière peut exiger une stratégie transactionnelle beaucoup plus stricte.
Celery fournit l’infrastructure d’exécution, mais la sémantique métier doit venir de votre application.
Pattern : un job parent et plusieurs tâches enfants
Pour un import :
ImportJob 42
|
+-- Batch 1
+-- Batch 2
+-- Batch 3
+-- Batch 4
Chaque batch peut être une tâche :
@shared_task
def process_batch(job_id, batch_id):
...
Puis une tâche finale :
@shared_task
def finalize_job(job_id):
...
Vous pouvez utiliser les primitives group et chord de Celery pour orchestrer ce workflow.
Les tâches de nettoyage
Celery est particulièrement utile pour les opérations de maintenance :
supprimer fichiers temporaires
archiver anciennes données
nettoyer sessions
purger logs
recalculer statistiques
synchroniser cache
Par exemple :
@shared_task
def supprimer_fichiers_temporaires():
...
Puis avec Beat :
CELERY_BEAT_SCHEDULE = {
"cleanup-every-night": {
"task": "core.tasks.supprimer_fichiers_temporaires",
"schedule": crontab(
hour=2,
minute=30,
),
},
}
Cette architecture permet au serveur web de rester concentré sur les requêtes.
Faire attention aux tâches périodiques qui se chevauchent
Imaginez une tâche toutes les cinq minutes :
00:00 -> démarre
00:05 -> démarre encore
mais si la première prend :
7 minutes
vous avez :
00:00 Task A
00:05 Task B
00:07 Task A terminée
Les tâches se chevauchent.
Cela peut être voulu, mais souvent ce n’est pas le cas.
Il faut donc parfois introduire un verrou distribué.
Exemple de lock Redis conceptuel
Une stratégie possible consiste à utiliser un lock Redis :
from redis import Redis
redis_client = Redis.from_url(
settings.CELERY_BROKER_URL
)
@shared_task
def synchronisation():
lock = redis_client.lock(
"lock:synchronisation",
timeout=300,
)
if not lock.acquire(blocking=False):
return "already running"
try:
effectuer_synchronisation()
finally:
lock.release()
Il faut être extrêmement attentif aux timeouts du lock, aux crashs et au modèle de concurrence avant d’utiliser cette approche comme mécanisme critique.
Pour certains cas, une stratégie basée sur la base de données et des contraintes explicites peut être plus appropriée.
Monitoring métier en plus du monitoring Celery
Flower vous montre que :
Task -> SUCCESS
mais votre produit peut avoir besoin de savoir :
Import -> 75 %
ou :
Commande -> synchronisation fournisseur réussie
Le meilleur monitoring combine donc :
Infrastructure
+
Celery
+
Métier
Cela vous donne une vision réellement exploitable.
Une architecture de logs structurés
Vous pouvez par exemple produire :
logger.info(
"report_generation_started",
extra={
"report_id": report_id,
"task_id": self.request.id,
},
)
Puis :
logger.info(
"report_generation_completed",
extra={
"report_id": report_id,
"task_id": self.request.id,
},
)
Un système de logs centralisé peut ensuite rechercher :
report_id = 42
task_id = ...
Cette approche est très utile lorsqu’une tâche traverse plusieurs services.
Une tâche peut échouer après avoir effectué une partie du travail
C’est l’un des bugs les plus subtils.
Supposons :
@shared_task
def importer():
importer_ligne_1()
importer_ligne_2()
importer_ligne_3()
raise Exception()
Si vous retry :
importer()
les lignes 1 et 2 peuvent être traitées deux fois.
Une tâche « retryable » doit donc être conçue avec soin.
La bonne architecture peut être :
1 tâche = 1 élément idempotent
plutôt que :
1 énorme tâche = 100 000 opérations
Le principe du « petit job »
Une bonne tâche possède souvent :
entrée claire
sortie claire
durée raisonnable
effets limités
reprise possible
logs utiles
idempotence
Exemple :
@shared_task
def synchroniser_commande(order_id):
order = Order.objects.get(pk=order_id)
if order.synced_at:
return
synchronize(order)
order.synced_at = timezone.now()
order.save(
update_fields=["synced_at"]
)
Ce modèle est beaucoup plus facile à raisonner.
Quand Celery n’est pas nécessaire
Il existe des situations où Celery serait excessif.
Par exemple :
def additionner(a, b):
return a + b
n’a pas besoin d’une file.
Une opération triviale :
User.objects.filter(is_active=True).count()
n’a pas besoin de Celery si elle est rapide.
Le but n’est pas de transformer toute l’application en système distribué.
Le but est de déplacer les opérations qui bénéficient réellement de l’exécution différée.
Une règle pratique pour identifier les tâches
Une opération est un bon candidat lorsqu’elle est :
longue
coûteuse
non bloquante pour l’utilisateur
répétitive
planifiée
dépendante d’un service externe
facilement découpable
Exemples classiques :
envoi d’e-mails
génération de PDF
traitement d’images
imports
exports
webhooks
synchronisation API
calculs lourds
nettoyage
notifications
recalculs
rapports
Celery et les webhooks
Les webhooks sont un autre excellent usage.
Votre endpoint :
@api_view(["POST"])
def webhook(request):
payload = request.data
traiter_webhook.delay(
request.headers.get("X-Event-ID")
)
return Response(
{"status": "accepted"},
status=202,
)
Cela permet de répondre rapidement au fournisseur externe.
Mais attention : le webhook peut lui aussi être envoyé deux fois.
Il faut alors utiliser une clé d’événement unique :
class WebhookEvent(models.Model):
event_id = models.CharField(
max_length=255,
unique=True,
)
La contrainte unique devient votre protection.
Exemple robuste de traitement de webhook
@shared_task
def traiter_webhook(event_id):
event = WebhookEvent.objects.get(
event_id=event_id
)
if event.processed_at:
return
effectuer_action_metier(event)
event.processed_at = timezone.now()
event.save(
update_fields=["processed_at"]
)
Ce modèle résiste beaucoup mieux aux doublons.
Définir une stratégie d’échec
Pour chaque tâche importante, écrivez mentalement :
Que se passe-t-il si la tâche échoue ?
Que se passe-t-il si elle échoue cinq fois ?
Que se passe-t-il si elle est exécutée deux fois ?
Que se passe-t-il si Redis tombe ?
Que se passe-t-il si la base tombe ?
Que se passe-t-il si l’API externe répond lentement ?
Que se passe-t-il si le worker est arrêté ?
Cette série de questions transforme une simple « tâche Celery » en composant fiable.
La philosophie des retries
Un retry n’est pas une manière de cacher les erreurs.
Un retry est une stratégie de récupération.
Si votre code contient :
@shared_task(
autoretry_for=(Exception,),
retry_kwargs={"max_retries": 100},
)
vous avez probablement masqué un problème plutôt que résolu la fiabilité.
Préférez :
erreur transitoire -> retry
erreur permanente -> failure
bug logiciel -> corriger
donnée invalide -> validation
rate limit -> backoff
Exemple de configuration de tâches robuste
Voici un exemple réaliste :
from celery import shared_task
from requests.exceptions import (
ConnectionError,
Timeout,
)
@shared_task(
bind=True,
max_retries=5,
acks_late=True,
)
def synchroniser_client(self, client_id):
client = Client.objects.get(pk=client_id)
try:
response = requests.get(
client.remote_url,
timeout=15,
)
response.raise_for_status()
except (ConnectionError, Timeout) as exc:
raise self.retry(
exc=exc,
countdown=60,
)
data = response.json()
appliquer_synchronisation(
client,
data,
)
Il faut toutefois considérer acks_late avec beaucoup de prudence et comprendre les conséquences en cas de crash ou de redélivrance. L’utilisation de l’acknowledgement tardif renforce l’importance de concevoir les tâches de manière idempotente.
Pourquoi acks_late peut changer la manière de penser les tâches
Avec une stratégie d’acknowledgement tardive, vous acceptez que le système puisse redonner une tâche au broker lorsqu’un worker disparaît avant la fin.
C’est potentiellement excellent pour la fiabilité.
Mais cela implique :
tâche exécutable plusieurs fois
Il faut donc penser :
idempotence
atomicité
transactions
verrous
contraintes uniques
La fiabilité du système ne vient jamais d’une seule option Celery. Elle vient de la combinaison de plusieurs couches.
Dépannage : Redis Connection Error
Erreur fréquente :
ConnectionError: Error connecting to Redis
Vérifiez :
redis-cli ping
Puis :
systemctl status redis-server
Ensuite vérifiez l’URL :
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
Si Django tourne dans Docker et Redis tourne dans un autre conteneur, 127.0.0.1 est généralement incorrect depuis le conteneur Django.
Dans Docker Compose :
CELERY_BROKER_URL = "redis://redis:6379/0"
Ici :
redis
est le nom du service Docker.
Dépannage : Celery ne trouve pas les tâches
Si le worker démarre mais n’affiche pas vos tâches, vérifiez :
app.autodiscover_tasks()
et la présence de :
tasks.py
dans l’application.
Vérifiez aussi que l’application est bien dans :
INSTALLED_APPS
Par exemple :
INSTALLED_APPS = [
...
"commandes",
]
Puis redémarrez le worker.
Le processus worker n’est pas automatiquement rechargé lorsqu’un fichier Python change.
Dépannage : changement de code non pris en compte
Si vous modifiez :
tasks.py
le worker déjà lancé continue avec l’ancien code.
Redémarrez-le :
pkill -f 'celery worker'
ou utilisez le mécanisme de supervision prévu dans votre environnement.
En développement, certaines solutions permettent un rechargement automatique, mais en production il est préférable de redémarrer proprement les workers.
Dépannage : tâche bloquée à PENDING
Une tâche PENDING peut avoir plusieurs causes.
Par exemple :
worker arrêté
broker inaccessible
mauvaise application Celery
mauvaise configuration
résultat non disponible
task id incorrect
Commencez par vérifier le worker :
celery -A monprojet inspect ping
Vous pouvez également inspecter les workers :
celery -A monprojet inspect active
et :
celery -A monprojet inspect registered
Le nom de la commande et les possibilités exactes dépendent de votre version et du setup, mais l’idée générale est de vérifier que le worker est vivant et qu’il connaît bien vos tâches.
Dépannage : la tâche échoue avec une erreur de sérialisation
Un exemple :
Object of type User is not JSON serializable
La solution consiste généralement à transmettre des primitives :
task.delay(user.id)
plutôt qu’un modèle complet :
task.delay(user)
Préférez :
int
str
float
bool
liste simple
dict JSON-compatible
et des identifiants pour les objets lourds.
Dépannage : Redis manque de mémoire
Si Redis est utilisé comme broker + backend + cache + sessions, vous avez probablement regroupé trop de responsabilités dans le même service.
Surveillez :
memory usage
evictions
fragmentation
number of keys
result TTL
Puis demandez-vous :
Quelles données doivent réellement être dans Redis ?
Les résultats Celery expirables et les caches temporaires ne devraient pas forcément occuper toute votre mémoire.
Dépannage : tâches très lentes
Commencez par mesurer.
Ajoutez :
import time
start = time.monotonic()
...
elapsed = time.monotonic() - start
logger.info(
"Task finished in %.2f seconds",
elapsed,
)
Puis identifiez si le temps est consommé par :
SQL
HTTP
CPU
disk
serialization
Redis
Une optimisation efficace commence presque toujours par une mesure.
Optimiser le nombre de requêtes SQL
Django Debug Toolbar est excellent pour les requêtes web, mais les tâches Celery doivent également être analysées.
Vous pouvez regarder :
from django.db import connection
print(len(connection.queries))
à des fins de diagnostic en développement.
Mais pour les tâches importantes, utilisez plutôt des outils de profiling ou d’observabilité adaptés.
Ne logguez pas des millions de lignes
Une tâche qui traite :
1 000 000 éléments
ne doit pas écrire :
logger.info("processed item %s", item.id)
à chaque itération en production.
Vous allez générer un volume énorme de logs.
Préférez des checkpoints :
if index % 1000 == 0:
logger.info(
"Processed %s items",
index,
)
Cela donne une visibilité suffisante sans transformer les logs en base de données secondaire.
Nettoyer les résultats Celery
Si vous stockez des résultats dans Redis, prévoyez leur expiration :
CELERY_RESULT_EXPIRES = 3600
Et pour les tâches qui n’ont aucune valeur de retour utile, ne retournez pas des structures gigantesques.
Préférez :
return {"success": True}
ou :
return None
selon vos besoins.
Architecture complète d’une application de production
Voici un exemple de séparation raisonnable :
INTERNET
|
v
Load Balancer
|
+----------+----------+
| |
v v
Django 1 Django 2
| |
+----------+----------+
|
v
Redis
|
+-------------+-------------+
| | |
v v v
worker-web worker-heavy worker-email
| | |
+-------------+-------------+
|
v
PostgreSQL
Puis :
Celery Beat
|
v
Redis
|
v
Workers
Le gros avantage est la capacité à scaler chaque composant séparément.
Si les e-mails explosent :
+ workers email
Si les rapports explosent :
+ workers reports
Le serveur Django n’a pas nécessairement besoin d’être multiplié dans les mêmes proportions.
Scaling horizontal
Vous pouvez avoir :
Worker machine A
Worker machine B
Worker machine C
tous connectés au même broker.
Cela donne :
Redis
/ | \
/ | \
v v v
W1 W2 W3
Celery est conçu précisément autour de ce modèle de workers répartis.
Il faut toutefois faire attention aux dépendances :
database capacity
Redis capacity
external API limits
network bandwidth
storage
Scaler les workers ne supprime pas les autres limites.
Queue prioritaire pour les tâches importantes
Vous pouvez séparer :
critical
default
low
Par exemple :
critical -> notifications critiques
default -> traitement normal
low -> rapports lourds
Puis lancer des workers spécialisés :
celery -A monprojet worker -Q critical,default --concurrency=4
et :
celery -A monprojet worker -Q low --concurrency=2
Cela vous permet d’éviter qu’une avalanche de rapports lourds bloque complètement les tâches importantes.
Le piège des tâches qui appellent d’autres tâches
Vous pouvez faire :
@shared_task
def tache_a():
tache_b.delay()
Cela fonctionne, mais il faut éviter de créer des chaînes implicites difficiles à suivre.
Préférez souvent les primitives explicites :
chain(
tache_a.s(),
tache_b.s(),
)
ou un orchestrateur métier clair.
L’objectif est que l’architecture soit lisible lorsque quelqu’un devra la diagnostiquer six mois plus tard.
Documentation interne des tâches
Pour chaque tâche critique, documentez :
Nom
Entrées
Sortie
Queue
Durée moyenne
Retry
Timeout
Idempotence
Effets secondaires
Dépendances
Exemple :
Task: synchroniser_commande
Input:
order_id: int
Queue:
orders
Retry:
Timeout, ConnectionError
Max retries:
5
Average duration:
2.4 s
Idempotent:
Oui
External dependencies:
Payment API
Business side effects:
update payment status
Cette documentation peut sembler excessivement formelle au début, mais elle devient précieuse dès que l’équipe grandit.
Une convention de nommage utile
Préférez :
accounts.tasks.send_welcome_email
reports.tasks.generate_monthly_report
orders.tasks.sync_order
media.tasks.generate_thumbnail
plutôt que des noms génériques :
process
handle
run
execute
Un nom précis facilite énormément le monitoring.
Organisation des fichiers tasks.py
Dans les petits projets :
app/
tasks.py
suffit.
Dans les projets plus grands, vous pouvez organiser :
app/
tasks/
__init__.py
emails.py
reports.py
imports.py
synchronization.py
Puis exposer les tâches.
Le but est d’éviter un fichier :
tasks.py
de 5000 lignes.
Exemple d’organisation modulaire
orders/
├── tasks/
│ ├── __init__.py
│ ├── notifications.py
│ ├── synchronization.py
│ └── exports.py
├── services/
│ ├── payment.py
│ └── shipping.py
Puis :
# orders/tasks/notifications.py
@shared_task
def send_order_confirmation(order_id):
...
Cette séparation est particulièrement utile lorsque la logique métier commence à devenir complexe.
Ne mettez pas toute la logique métier dans le décorateur Celery
Évitez :
@shared_task
def task():
# 500 lignes de logique
Le décorateur devrait principalement transformer une fonction métier en tâche.
Préférez :
@shared_task
def synchroniser_commande(order_id):
return OrderService().synchronize(order_id)
Vous pouvez ensuite tester OrderService indépendamment.
Service métier + Celery
Exemple :
class OrderService:
def synchronize(self, order_id):
order = Order.objects.get(pk=order_id)
...
Puis :
@shared_task
def synchronize_order(order_id):
return OrderService().synchronize(order_id)
Cela facilite :
unit tests
réutilisation
API
CLI
Celery
Vous ne rendez pas votre logique métier dépendante de Celery.
Exécuter les tâches depuis une commande Django
Parfois vous souhaitez réutiliser le même service depuis :
python manage.py synchronize_orders
et :
synchronize_order.delay(order.id)
Le service métier devient alors le point commun.
C’est une excellente façon de construire une architecture qui n’est pas prisonnière d’un seul mécanisme d’exécution.
Celery n’est pas une base de données métier
Une erreur conceptuelle consiste à utiliser le backend de résultats comme journal métier.
Par exemple :
Celery result = vérité de l’application
Ce n’est généralement pas une bonne idée.
Pour une commande :
Order.status
Order.paid_at
Order.shipped_at
doivent vivre dans vos modèles métier.
Celery orchestre l’exécution.
Celery n’est pas non plus un cache
Redis peut servir de cache Django et de broker Celery, mais cela ne signifie pas que tout doit partager exactement le même usage.
Si Redis est soudain vidé :
cache -> peut être régénéré
result -> peut être perdue selon la configuration
broker messages -> impact potentiellement critique
Il faut donc connaître la valeur des différentes données.
Les configurations modernes de Celery
La documentation actuelle de Celery continue de privilégier les noms de configuration modernes en minuscules côté API Celery, tout en documentant le mapping Django avec le préfixe CELERY_. Elle avertit également que les anciennes configurations finiront par être supprimées dans une future version majeure.
Dans Django, on utilise donc couramment :
CELERY_BROKER_URL = ...
CELERY_RESULT_BACKEND = ...
CELERY_TASK_SERIALIZER = ...
avec :
app.config_from_object(
"django.conf:settings",
namespace="CELERY",
)
Cette convention est propre et claire.
Un exemple complet minimal
settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "Africa/Casablanca"
celery.py
import os
from celery import Celery
os.environ.setdefault(
"DJANGO_SETTINGS_MODULE",
"monprojet.settings",
)
app = Celery("monprojet")
app.config_from_object(
"django.conf:settings",
namespace="CELERY",
)
app.autodiscover_tasks()
init.py
from .celery import app as celery_app
__all__ = ("celery_app",)
tasks.py
from celery import shared_task
@shared_task
def additionner(a, b):
return a + b
démarrage
redis-server
Puis :
celery -A monprojet worker -l info
Puis dans Django :
from commandes.tasks import additionner
result = additionner.delay(5, 7)
print(result.id)
Le worker doit exécuter :
additionner(5, 7)
et produire :
12
Un exemple complet avec modèle Django
models.py
from django.db import models
class Report(models.Model):
name = models.CharField(max_length=255)
status = models.CharField(
max_length=30,
default="pending",
)
result = models.TextField(
blank=True,
null=True,
)
created_at = models.DateTimeField(
auto_now_add=True,
)
updated_at = models.DateTimeField(
auto_now=True,
)
tasks.py
from celery import shared_task
from .models import Report
@shared_task(bind=True)
def generate_report(self, report_id):
report = Report.objects.get(
pk=report_id
)
report.status = "processing"
report.save(
update_fields=["status"]
)
try:
result = create_report(report)
report.result = result
report.status = "completed"
report.save(
update_fields=[
"result",
"status",
]
)
except Exception:
report.status = "failed"
report.save(
update_fields=["status"]
)
raise
views.py
from django.db import transaction
from django.http import JsonResponse
from .models import Report
from .tasks import generate_report
def create_report(request):
report = Report.objects.create(
name=request.POST["name"]
)
transaction.on_commit(
lambda: generate_report.delay(
report.id
)
)
return JsonResponse(
{
"id": report.id,
"status": report.status,
},
status=202,
)
Ce modèle contient déjà plusieurs bonnes pratiques :
tâche asynchrone
identifiant léger
transaction.on_commit
état métier
gestion d’erreur
HTTP 202
Une version encore plus robuste avec retries
from celery import shared_task
from requests.exceptions import (
ConnectionError,
Timeout,
)
from .models import Report
@shared_task(
bind=True,
max_retries=5,
soft_time_limit=900,
)
def generate_report(self, report_id):
report = Report.objects.get(
pk=report_id
)
report.status = "processing"
report.save(
update_fields=["status"]
)
try:
result = external_generate_report(
report
)
except (ConnectionError, Timeout) as exc:
raise self.retry(
exc=exc,
countdown=60,
)
except Exception:
report.status = "failed"
report.save(
update_fields=["status"]
)
raise
report.result = result
report.status = "completed"
report.save(
update_fields=[
"result",
"status",
]
)
La robustesse ne vient pas d’un seul morceau du code, mais de la combinaison :
timeout
retry
état métier
transaction
identifiant
idempotence
monitoring
Que faire quand Redis est indisponible ?
Supposons :
Django -> Redis X
Alors :
task.delay(...)
peut échouer immédiatement.
C’est une réalité importante : mettre une tâche en arrière-plan ne signifie pas que la remise du message est toujours disponible.
Pour les opérations critiques, vous pouvez avoir besoin d’une stratégie supplémentaire :
transaction + outbox
Pattern Outbox pour les systèmes critiques
L’idée générale est de stocker dans la base une intention de traitement dans la même transaction que la modification métier.
Exemple :
class OutboxEvent(models.Model):
event_type = models.CharField(
max_length=100
)
payload = models.JSONField()
published = models.BooleanField(
default=False
)
created_at = models.DateTimeField(
auto_now_add=True
)
Dans la même transaction :
with transaction.atomic():
order = create_order()
OutboxEvent.objects.create(
event_type="order.created",
payload={
"order_id": order.id,
},
)
Puis un worker ou un processus de publication peut lire les événements non publiés.
Ce modèle est plus complexe, mais il est particulièrement utile lorsqu’il est critique de ne pas perdre un événement métier entre la base et le broker.
Quand envisager l’Outbox Pattern
L’Outbox devient particulièrement intéressant lorsque :
création en DB
+
publication message
doivent être cohérentes.
Sans outbox :
DB commit
|
X
Redis indisponible
L’état métier existe mais le message n’a jamais été publié.
Avec outbox :
DB commit
|
+-- métier
+-- outbox event
Le message peut être publié ensuite.
Ce pattern dépasse la configuration minimale de Celery, mais il illustre bien jusqu’où l’architecture doit parfois aller lorsque la fiabilité devient une priorité.
L’importance de la simplicité
Tout ce que nous avons vu peut sembler très sophistiqué.
Pour une petite application, vous pouvez commencer avec :
Django
Redis
Celery worker
et seulement quelques tâches.
Puis ajouter progressivement :
Celery Beat
multiple queues
Flower
retries
timeouts
job models
monitoring
outbox
L’erreur serait de construire dès le jour 1 une plateforme distribuée gigantesque pour une application qui traite 50 tâches par jour.
L’autre erreur serait de rester avec une architecture naïve alors que l’application traite 500 000 tâches par heure.
L’architecture doit suivre les besoins.
Une feuille de route réaliste
Pour un nouveau projet Django :
Étape 1
Installer Redis
Étape 2
Installer Celery
Étape 3
Créer celery.py
Étape 4
Configurer broker
Étape 5
Créer une tâche simple
Étape 6
Démarrer le worker
Étape 7
Appeler delay()
Étape 8
Ajouter gestion d’erreurs
Étape 9
Ajouter transaction.on_commit
Étape 10
Ajouter monitoring
Étape 11
Découper les queues
Étape 12
Planifier les tâches
Étape 13
Optimiser la production
Cette progression vous permet de construire une solution sans vous noyer immédiatement dans les options avancées.
Checklist avant la production
Avant de déployer Celery, vérifiez au minimum :
[ ] Redis accessible uniquement depuis les réseaux nécessaires
[ ] secrets stockés en variables d’environnement
[ ] worker géré par systemd, Supervisor ou orchestrateur
[ ] Beat géré séparément
[ ] logs collectés
[ ] timeouts configurés
[ ] retries définis pour les erreurs transitoires
[ ] tâches critiques idempotentes
[ ] appels externes avec timeout
[ ] résultats Celery avec expiration
[ ] queues séparées si nécessaire
[ ] monitoring
[ ] alertes
[ ] stratégie de redémarrage
[ ] stratégie de déploiement compatible avec les workers
Les erreurs les plus fréquentes avec Celery et Django
La première erreur est d’oublier de démarrer le worker.
Vous avez :
task.delay()
mais aucun worker.
Redis reçoit le message et celui-ci reste en attente.
La deuxième erreur est de lancer le worker dans le mauvais projet :
celery -A wrongproject worker
La troisième est d’utiliser :
localhost
depuis Docker alors que Redis est un autre service.
La quatrième est de passer des objets Django énormes aux tâches.
La cinquième est d’attendre le résultat avec :
result.get()
dans une requête web.
La sixième est de faire des retries infinis.
La septième est d’oublier transaction.on_commit() lorsque la tâche dépend d’une donnée transactionnelle fraîchement créée.
La huitième est de supposer qu’une tâche ne sera jamais exécutée deux fois.
La neuvième est de transformer Celery Beat en système de logique métier sans contrôle du chevauchement.
La dixième est de ne jamais surveiller le backlog.
Le vrai changement de mentalité
Au départ, on pense :
fonction -> résultat
Avec Celery, il faut penser :
commande -> message -> attente -> exécution -> état -> résultat
Cette différence semble subtile mais elle change toute la manière de concevoir le code.
Une fonction classique peut être pensée comme :
result = do_something()
Une tâche distribuée doit être pensée comme :
task_id = do_something.delay()
puis :
Que fait-on si elle échoue ?
Qui est responsable ?
Comment vérifier son état ?
Comment savoir si elle a été exécutée deux fois ?
Combien de temps peut-elle durer ?
Que se passe-t-il si le worker redémarre ?
C’est ce passage d’un modèle synchrone à un modèle distribué qui constitue la vraie courbe d’apprentissage de Celery.
Un exemple complet d’architecture pour une application SaaS
Imaginez une plateforme SaaS Django qui possède :
users
subscriptions
invoices
emails
reports
notifications
imports
exports
webhooks
Vous pourriez utiliser :
Redis
|
+-- email
+-- reports
+-- imports
+-- exports
+-- webhooks
Avec :
Worker Email x4
Worker Reports x2
Worker Imports x2
Worker Webhooks x2
et :
Celery Beat
pour :
factures
renouvellements
nettoyages
statistiques
Le serveur HTTP reste spécialisé dans :
API
HTML
authentification
CRUD
alors que les workers se concentrent sur :
travail différé
traitements lourds
intégrations
automatisation
Cette séparation améliore souvent la lisibilité et la scalabilité du système.
Ce qu’il faut retenir sur Django et Celery en 2026
En 2026, Celery 5.6 constitue la branche stable documentée et son intégration Django reste activement documentée. Redis est toujours un broker stable pris en charge et peut également servir de backend de résultats.
Dans le même temps, Django moderne dispose désormais de sa propre API Tasks, introduite avec Django 6.0, ce qui signifie que l’écosystème Django dispose aujourd’hui de plusieurs niveaux d’abstraction pour les tâches de fond. Cette API native définit le contrat des tâches mais laisse l’exécution à une infrastructure externe, tandis que Celery fournit un système complet de workers, de brokers, de planification et d’orchestration.
Pour de nombreux projets Django nécessitant une file d’attente mature, Redis + Celery reste donc une combinaison très pertinente. Le choix doit néanmoins se faire en fonction du volume, de la durabilité recherchée, des besoins d’orchestration et de l’infrastructure globale.
Conclusion
Exécuter des tâches en arrière-plan avec Celery et Redis dans Django n’est pas seulement une question d’ajouter un package et de taper delay(). La commande est facile. La partie réellement intéressante commence ensuite.
Il faut comprendre que Django reçoit la requête, Celery représente le moteur d’exécution, Redis transporte les messages et peut conserver les résultats, tandis que les workers font le travail réel. Une fois cette architecture comprise, de nombreuses fonctionnalités deviennent beaucoup plus faciles à concevoir. Un e-mail qui prenait trois secondes devient une tâche. Un rapport PDF qui prenait trente secondes devient un job asynchrone. Un import de plusieurs milliers de lignes peut être découpé en dizaines ou centaines de tâches. Une synchronisation avec une API externe peut être protégée par des retries et des timeouts. Une maintenance nocturne peut être planifiée avec Celery Beat.
Mais la vraie maturité ne consiste pas à déplacer toutes les fonctions lentes vers Celery. Elle consiste à construire des tâches que l’on peut réellement faire fonctionner dans un environnement distribué. Cela implique de penser à l’idempotence, aux retries, aux erreurs transitoires, aux transactions Django, aux timeouts, à la taille des messages, à la mémoire Redis, aux files spécialisées, au monitoring, aux redéliveries et aux déploiements.
Le principe le plus important à retenir est probablement celui-ci : une tâche en arrière-plan doit être considérée comme un travail qui peut commencer plus tard, échouer, être relancé, être retardé ou même être exécuté plusieurs fois. Une fois que votre code est conçu avec cette réalité en tête, Celery cesse d’être une simple bibliothèque et devient un véritable composant architectural.
Pour commencer simplement, vous pouvez garder une configuration très légère :
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
puis :
@shared_task
def ma_tache(object_id):
...
et :
ma_tache.delay(object_id)
Mais au fur et à mesure que votre application grandit, vous pourrez ajouter :
transaction.on_commit()
retries
timeouts
rate limits
queues
Celery Beat
Flower
job tracking
idempotency
distributed locks
outbox
observability
C’est cette progression qui permet de conserver une architecture Django simple au début tout en gardant la possibilité de passer à une infrastructure beaucoup plus robuste.
Et il y a quelque chose de très satisfaisant lorsque tout fonctionne correctement : vous cliquez sur « Générer le rapport », Django répond immédiatement, Redis reçoit le message, un worker démarre quelques millisecondes plus tard, le fichier est généré sans bloquer le serveur web, puis votre interface affiche tranquillement « Rapport prêt ». À ce moment-là, la magie semble presque invisible. En réalité, elle repose sur une séparation de responsabilités soigneusement pensée.
Celery et Redis ne sont donc pas seulement des outils pour « faire des tâches en arrière-plan ». Ils permettent surtout de repenser la façon dont une application Django travaille. Au lieu de tout exécuter au même endroit, au même moment et dans la même requête, on peut répartir le travail, le contrôler, le surveiller, le réessayer et le faire évoluer indépendamment. C’est précisément cette capacité qui rend les files de tâches si puissantes dans les applications modernes.
Et comme souvent en ingénierie logicielle, la meilleure solution n’est pas celle qui possède le plus de workers ou le plus de configuration. C’est celle qui rend le système suffisamment rapide pour l’utilisateur, suffisamment simple pour l’équipe et suffisamment robuste pour survivre aux problèmes réels qui finiront forcément par arriver.