Celery et Redis pour les tâches asynchrones en Python

sudo apt update && sudo apt install redis-server

Démarrez le service Redis et vérifiez qu'il est actif :

sudo systemctl start redis-server
sudo systemctl enable redis-server

Redis écoute par défaut sur le port TCP 6379. Vérifiez cela avec :

redis-cli ping

La sortie doit être `PONG`.

2. Installez Celery avec le support Redis

Celery nécessite un package séparé pour l'intégration Redis. Installez `celery[redis]` afin que toutes les dépendances pour le support Redis soient incluses :

pip install "celery[redis]"

Sans ce package supplémentaire, Celery ne peut pas se connecter à Redis, même si le serveur Redis est en cours d'exécution.

3. Configurez l'application Celery

Créez une application Celery qui utilise Redis comme courtier de messages (message broker) et comme backend de résultats. Créez un fichier `celery_app.py` :

from celery import Celery</p><p>app = Celery(
    'tasks',
    broker='redis://localhost:6379/10',
    backend='redis://localhost:6379/11'
)</p><p>app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Europe/Brussels',
    enable_utc=True,
)

Le broker et le backend pointent vers les bases de données Redis 10 et 11 respectivement. Ce n'est pas une configuration obligatoire mais une manière courante d'isoler les flux de données. Vous pouvez aussi utiliser la base de données 0 avec `redis://localhost:6379/0`.

4. Définissez une tâche asynchrone

Les tâches Celery sont des fonctions Python ordinaires avec le décorateur `@app.task`. Créez `tasks.py` :

from celery_app import app</p><p>@app.task
def send_email(recipient: str, subject: str, body: str) -> dict:
    # Simulez une opération de longue durée
    import time
    time.sleep(5)
    
    # Ici, vous enverriez réellement un e-mail
    return {
        'recipient': recipient,
        'subject': subject,
        'status': 'sent'
    }</p><p>@app.task
def process_upload(file_path: str) -> dict:
    # Traitez un fichier téléversé
    with open(file_path, 'rb') as f:
        data = f.read()
    
    # Faites quelque chose avec les données
    result = {'bytes_processed': len(data)}
    
    return result

Le décorateur permet à Celery de reconnaître la fonction comme une tâche asynchrone qui peut être exécutée par un worker.

5. Démarrez le worker Celery

Le worker s'exécute dans un processus séparé, indépendant de l'application principale. Démarrez le worker avec :

celery -A tasks worker --loglevel=info

Le worker écoute les messages dans la file Redis et traite les tâches dès qu'elles arrivent. C'est essentiel pour le fonctionnement asynchrone : le serveur web continue de répondre aux requêtes des utilisateurs, tandis que le worker gère les tâches lourdes en arrière-plan.

6. Envoyez des tâches depuis votre application

Appelez les tâches avec `.delay()` ou `.apply_async()`. Créez `main.py` :

from tasks import send_email, process_upload
<h2>Appel simple avec .delay()</h2>
result = send_email.delay(
    'user@example.com',
    'Bienvenue',
    'Merci pour votre inscription'
)
<h2>Appel avancé avec .apply_async()</h2>
result = process_upload.apply_async(
    args=['/tmp/fichier.pdf'],
    countdown=10,  # démarre après 10 secondes
    retry=True,
    retry_policy={
        'max_retries': 3,
        'interval_start': 0,
        'interval_step': 0.2,
        'interval_max': 0.2,
    }
)

`.delay()` est une forme abrégée de `.apply_async()` et n'accepte que des arguments positionnels. `.apply_async()` offre plus de contrôle comme les délais, les nouvelles tentatives, les callbacks et la sélection de file.

7. Récupérez les résultats des tâches

Utilisez l'objet résultat pour interroger le statut et le résultat d'une tâche. Redis sert ici de backend de résultats où les résultats sont temporairement stockés :

from tasks import send_email</p><p>result = send_email.delay('user@example.com', 'Test', 'Corps')
<h2>Vérifiez si la tâche est terminée</h2>
print(f"Tâche terminée : {result.ready()}")
<h2>Attendez le résultat (bloque jusqu'à ce que la tâche soit terminée)</h2>
result_data = result.get(timeout=10)
print(f"Résultat : {result_data}")
<h2>Récupérez l'ID de la tâche</h2>
task_id = result.id
print(f"ID de la tâche : {task_id}")

Cette fonctionnalité est utile lorsqu'un utilisateur doit être informé de la progression d'une opération de longue durée.

8. Planifiez des tâches périodiques

Celery peut aussi exécuter des jobs périodiques via `celery beat`. Ajoutez cela à `celery_app.py` :

from celery.schedules import crontab</p><p>app.conf.beat_schedule = {
    'envoi-rapports-quotidiens': {
        'task': 'tasks.send_email',
        'schedule': crontab(hour=8, minute=0),
        'args': ('admin@example.com', 'Rapport quotidien', 'Voici le rapport'),
    },
}

Démarrez le planificateur beat avec le worker :

celery -A tasks beat --loglevel=info

from tasks import send_email
result = send_email.delay('test@example.com', 'Test', 'Body')
print(result.ready())  # False si la tâche est encore en cours
print(result.get(timeout=10))  # Attend le résultat

broker='redis://localhost:6379/10',
backend='redis://localhost:6379/11'