django-celery

Modèles de Tâches Asynchrones Django + Celery

Modèles de production pour le traitement de tâches asynchrones en arrière-plan dans Django en utilisant Celery avec Redis ou RabbitMQ.

Quand Activer

• Ajouter des travaux d'arrière-plan ou du traitement asynchrone à une application Django
• Implémenter des tâches périodiques/planifiées
• Décharger les opérations lentes (email, génération de PDF, appels API) du cycle de requête
• Mettre en place Celery Beat pour une planification de type cron
• Déboguer les échecs de tâches, les tentatives ou les files d'attente en retard
• Écrire des tests pour les tâches Celery

Configuration du Projet

Installation

pip install 'celery[redis]' django-celery-results django-celery-beat

celery.py — Point d'Entrée de l'Application

# config/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.development')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()  # Découvre les fichiers tasks.py dans chaque INSTALLED_APP

@app.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Requête : {self.request!r}')

# config/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

Paramètres Django

# config/settings/base.py

# Broker (Redis recommandé pour la production)
CELERY_BROKER_URL = env('CELERY_BROKER_URL', default='redis://localhost:6379/0')
CELERY_RESULT_BACKEND = env('CELERY_RESULT_BACKEND', default='django-db')

# Sérialisation
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# Comportement des tâches
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60        # Limite dure : 30 min
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60   # Limite souple : envoie SoftTimeLimitExceeded
CELERY_WORKER_PREFETCH_MULTIPLIER = 1   # Empêche le worker d'accaparer les longues tâches
CELERY_TASK_ACKS_LATE = True            # Remet en file d'attente en cas de crash du worker

# Persistance des résultats
CELERY_RESULT_EXPIRES = 60 * 60 * 24   # Conserver les résultats 24 heures

# Planificateur Beat (pour les tâches périodiques)
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# Applications installées
INSTALLED_APPS += [
    'django_celery_results',
    'django_celery_beat',
]

Lancement des Workers

# Démarrer le worker (développement)
celery -A config worker --loglevel=info

# Démarrer le planificateur Beat (tâches périodiques)
celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler

# Worker + Beat combinés (développement uniquement, jamais en production)
celery -A config worker --beat --loglevel=info

# Production : plusieurs workers avec concurrence
celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority

Modèles de Conception de Tâches

Tâche Basique

# apps/notifications/tasks.py
from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(name='notifications.send_welcome_email')
def send_welcome_email(user_id: int) -> None:
    """Envoie un email de bienvenue à un utilisateur nouvellement inscrit."""
    from apps.users.models import User
    from apps.notifications.services import EmailService

    try:
        user = User.objects.get(pk=user_id)
    except User.DoesNotExist:
        logger.warning('send_welcome_email : utilisateur %s introuvable', user_id)
        return  # Idempotent — ne pas lever d'exception, la tâche est déjà impossible à accomplir

    EmailService.send_welcome(user)
    logger.info('Email de bienvenue envoyé à l\'utilisateur %s', user_id)

Tâche avec Tentatives

@shared_task(
    bind=True,
    name='integrations.sync_to_crm',
    max_retries=5,
    default_retry_delay=60,       # secondes avant la première tentative
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,           # backoff exponentiel
    retry_backoff_max=600,        # plafonné à 10 minutes
    retry_jitter=True,            # aléatoire pour éviter l'effet de troupeau
)
def sync_contact_to_crm(self, contact_id: int) -> dict:
    """Synchronise un contact vers un CRM externe avec tentatives en cas d'échec transitoire."""
    from apps.crm.services import CRMClient

    try:
        result = CRMClient().sync(contact_id)
        return result
    except CRMClient.RateLimitError as exc:
        # Délai de tentative spécifique à partir de l'en-tête de réponse
        raise self.retry(exc=exc, countdown=int(exc.retry_after))

Modèle de Tâche Idempotente

Concevez les tâches pour qu'elles puissent être exécutées plusieurs fois sans effet de bord avec les mêmes entrées :

@shared_task(name='orders.mark_shipped')
def mark_order_shipped(order_id: int, tracking_number: str) -> None:
    """Marque une commande comme expédiée — peut être exécutée plusieurs fois sans danger."""
    from apps.orders.models import Order

    updated = Order.objects.filter(
        pk=order_id,
        status=Order.Status.PROCESSING,    # Garde : ne mettre à jour que si pas déjà expédiée
    ).update(
        status=Order.Status.SHIPPED,
        tracking_number=tracking_number,
    )

    if not updated:
        logger.info('mark_order_shipped : commande %s déjà expédiée ou introuvable', order_id)

Tâche avec Limite de Temps Souple

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(
    bind=True,
    name='reports.generate_pdf',
    soft_time_limit=120,
    time_limit=150,
)
def generate_pdf_report(self, report_id: int) -> str:
    """Génère un rapport PDF avec une gestion gracieuse du dépassement de temps."""
    from apps.reports.services import PDFGenerator

    try:
        path = PDFGenerator.build(report_id)
        return path
    except SoftTimeLimitExceeded:
        # Nettoyer les fichiers partiels avant l'arrêt brutal
        PDFGenerator.cleanup(report_id)
        raise

Appel des Tâches

from datetime import timedelta
from django.utils import timezone

# Lancer et oublier (asynchrone)
send_welcome_email.delay(user.pk)

# Planifier dans le futur
send_reminder.apply_async(args=[user.pk], countdown=3600)  # Dans 1 heure
send_reminder.apply_async(args=[user.pk], eta=timezone.now() + timedelta(days=1))

# Appliquer avec routage de file d'attente
sync_contact_to_crm.apply_async(args=[contact.pk], queue='high_priority')

# Exécuter de manière synchrone (tests / débogage uniquement)
result = generate_pdf_report.apply(args=[report.pk])

Planification Beat (Tâches Périodiques)

Planification Définie dans le Code

# config/settings/base.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'users.cleanup_expired_sessions',
        'schedule': crontab(hour=2, minute=0),   # 2h du matin tous les jours
    },
    'sync-inventory': {
        'task': 'products.sync_inventory',
        'schedule': 60.0,                         # toutes les 60 secondes
    },
    'weekly-digest': {
        'task': 'notifications.send_weekly_digest',
        'schedule': crontab(day_of_week='monday', hour=8, minute=0),
    },
}

Planification Définie en Base de Données (via django-celery-beat)

# Gérer les tâches périodiques depuis l'admin Django ou le code
from django_celery_beat.models import PeriodicTask, CrontabSchedule
import json

schedule, _ = CrontabSchedule.objects.get_or_create(
    hour='*/6', minute='0',
    timezone='UTC',
)

PeriodicTask.objects.update_or_create(
    name='Synchroniser l\'inventaire toutes les 6 heures',
    defaults={
        'crontab': schedule,
        'task': 'products.sync_inventory',
        'args': json.dumps([]),
        'enabled': True,
    }
)

Canvas : Chaînage et Regroupement de Tâches

from celery import chain, group, chord

# Chaîne : exécuter les tâches séquentiellement, en passant les résultats
pipeline = chain(
    fetch_data.s(source_id),
    transform_data.s(),          # reçoit le résultat de fetch_data comme premier argument
    load_to_warehouse.s(),
)
pipeline.delay()

# Groupe : exécuter les tâches en parallèle
parallel = group(
    send_welcome_email.s(user_id)
    for user_id in new_user_ids
)
parallel.delay()

# Chord : tâches parallèles + callback quand toutes sont terminées
result = chord(
    group(process_chunk.s(chunk) for chunk in data_chunks),
    aggregate_results.s(),       # appelée avec la liste des résultats des chunks
)
result.delay()

Gestion des Erreurs et File d'Attente Morte

# apps/core/tasks.py
from celery.signals import task_failure

@task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    """Journalise tous les échecs de tâches dans Sentry / système d'alerte."""
    import sentry_sdk
    with sentry_sdk.new_scope() as scope:
        scope.set_context('celery', {
            'task': sender.name,
            'task_id': task_id,
            'args': args,
            'kwargs': kwargs,
        })
        sentry_sdk.capture_exception(exception)

# Acheminer les tâches en échec vers une file d'attente morte après le nombre maximum de tentatives
@shared_task(
    bind=True,
    max_retries=3,
    name='payments.charge_card',
)
def charge_card(self, order_id: int) -> None:
    from apps.payments.models import Order, FailedCharge

    try:
        _do_charge(order_id)
    except Exception as exc:
        if self.request.retries >= self.max_retries:
            # Persister dans une table de file d'attente morte pour examen manuel
            FailedCharge.objects.create(
                order_id=order_id,
                error=str(exc),
                task_id=self.request.id,
            )
            return  # Ne pas lever d'exception — la tâche a définitivement échoué
        raise self.retry(exc=exc)

Test des Tâches Celery

Tests Unitaires (Sans Broker)

# tests/test_tasks.py
import pytest
from unittest.mock import patch, MagicMock
from apps.notifications.tasks import send_welcome_email

class TestSendWelcomeEmail:

    @pytest.mark.django_db
    def test_sends_email_to_existing_user(self, user):
        with patch('apps.notifications.services.EmailService') as mock_email:
            send_welcome_email(user.pk)
            mock_email.send_welcome.assert_called_once_with(user)

    @pytest.mark.django_db
    def test_skips_missing_user_gracefully(self):
        """Ne doit pas lever d'exception si l'utilisateur est supprimé entre la mise en file et l'exécution."""
        send_welcome_email(99999)  # Utilisateur inexistant — ne doit pas lever d'exception

Tests d'Intégration avec CELERY_TASK_ALWAYS_EAGER

# config/settings/test.py
CELERY_TASK_ALWAYS_EAGER = True      # Exécuter les tâches de manière synchrone dans les tests
CELERY_TASK_EAGER_PROPAGATES = True  # Relancer les exceptions des tâches

# tests/test_integration.py
@pytest.mark.django_db
def test_registration_triggers_welcome_email(client):
    with patch('apps.notifications.services.EmailService') as mock_email:
        response = client.post('/api/users/', {
            'email': '[email protected]',
            'password': 'strongpass123',
        })

    assert response.status_code == 201
    mock_email.send_welcome.assert_called_once()

Test des Tentatives

@pytest.mark.django_db
def test_task_retries_on_connection_error():
    with patch('apps.crm.services.CRMClient.sync') as mock_sync:
        mock_sync.side_effect = ConnectionError('timeout')

        with pytest.raises(ConnectionError):
            sync_contact_to_crm.apply(args=[1], throw=True)

        assert mock_sync.call_count == 1  # Première tentative uniquement en mode eager

Surveillance

# Inspecter les workers et files d'attente actifs
celery -A config inspect active
celery -A config inspect stats
celery -A config inspect reserved

# Vérifier les longueurs de file d'attente (Redis)
redis-cli llen celery

# Flower : moniteur temps réel basé sur le web
pip install flower
celery -A config flower --port=5555

Anti-Patterns

# MAUVAIS : Passer des instances de modèle — elles peuvent être obsolètes au moment de l'exécution
send_welcome_email.delay(user)        # Ne jamais passer d'objets ORM
send_welcome_email.delay(user.pk)     # Toujours passer les clés primaires (PK)

# MAUVAIS : Appeler des tâches de manière synchrone dans les vues de production
result = generate_report.apply()      # Bloque le thread de la requête

# MAUVAIS : Tâche non idempotente sans garde-fou
@shared_task
def charge_and_fulfill(order_id):
    order.charge()     # Peut débiter deux fois si la tâche réessaye !
    order.fulfill()

# BON : Idempotent avec garde de statut
@shared_task
def charge_and_fulfill(order_id):
    order = Order.objects.select_for_update().get(pk=order_id)
    if order.status != Order.Status.PENDING:
        return  # Déjà traité
    order.charge()
    order.fulfill()

Checklist de Production

Vérification	Paramètre
Redémarrage du worker en cas de crash	Unité supervisord ou systemd
CELERY_TASK_ACKS_LATE = True	Remet en file d'attente les tâches en cas de crash du worker
CELERY_WORKER_PREFETCH_MULTIPLIER = 1	Distribution équitable des longues tâches
Files d'attente séparées par priorité	-Q default,high_priority,low_priority
CELERY_TASK_SOFT_TIME_LIMIT défini	Délai d'attente gracieux avant l'arrêt brutal
Intégration Sentry	Capturer tous les signaux task_failure
Flower ou autre moniteur	Visibilité sur les profondeurs de files d'attente
Beat exécuté sur un seul nœud uniquement	Empêche l'exécution en double des tâches planifiées

Compétences Associées

• django-patterns — ORM, couche service et structure de projet
• django-tdd — Test des modèles, vues et services Django
• python-testing — Configuration et fixtures pytest