L’infrastructure des tâches dans Django¶

New in Django 6.0.

Pour une application web, il y a souvent du code à exécuter en plus du processus de transformation des requêtes HTTP en réponses HTTP. Pour certaines fonctionnalités, il peut être avantageux d’exécuter du code en dehors du cycle de requête-réponse.

C’est là qu’entrent en jeu les tâches d’arrière-plan.

Les tâches d’arrière-plan peuvent décharger le cycle de requête-réponse pour exécuter du code en dehors de ce cycle, potentiellement à une date ultérieure. Cela permet de garder les requêtes rapides, de réduire la latence et d’amélioration le ressenti des utilisateurs. Par exemple, une utilisatrice ne devrait pas avoir besoin d’attendre qu’un courriel soit envoyé avant que sa page termine de charger.

L’infrastructure des tâches de Django facilite la définition et la mise en file d’attente de ce genre de travail. Mais elle ne fournit pas de mécanisme d’exécution de ces tâches. L’exécution réelle doit être gérée par une infrastructure extérieure à Django, telle qu’un processus ou service séparé. À ce titre, un moteur de tâches capable de les exécuter sur ce service externe doit être choisi et configuré.

Notions de base sur les tâches d’arrière-plan¶

Lorsque du code doit être exécuté en arrière-plan, Django crée une tâche Task, qui est stockée dans une file d’attente. Cette tâche contient toutes les métadonnées utiles à son exécution, de même qu’un identifiant unique pour que Django puisse plus tard obtenir son résultat.

Un processus de travail ira piocher de nouvelles tâches à exécuter dans la file d’attente. Lorsqu’une nouvelle tâches est ajoutée, le processus réserve la tâche, l’exécute, et enregistre son statut et son résultat dans la file d’attente. Ces processus s’exécutent en dehors du cycle de requête-réponse.

Configuration d’un moteur de tâches¶

Le moteur de tâches détermine comment et où les tâches sont stockées en vue de leur exécution et comment elles sont exécutées. Différents moteurs de tâches possèdent différentes caractéristiques et options de configuration, ce qui peut influer sur la performance et la fiabilité de votre application. Django est livré avec des moteurs intégrés, mais ceux-ci sont uniquement pour du développement ou des tests.

Django gère la définition des tâches, leur validation, mise en file d’attente, et gestion de leurs résultats, pas leur exécution, donc les configurations de production ont besoin d’un moteur ou d’un processus de lancement qui exécute réellement ce qui se trouve en attente. Les options possibles sont énumérées sur la page Community Ecosystem.

Les moteurs de tâches sont configurés par le réglage TASKS de votre fichier de réglages. Même si la majorité des applications n’auront besoin que d’un seul moteur, il est possible d’en avoir plusieurs.

Exécution immédiate¶

Il s’agit du moteur par défaut si aucun autre n’a été indiqué dans le fichier des réglages. Ce moteur ImmediateBackend exécute immédiatement les tâches en file d’attente, au lieu de le faire en arrière-plan. Cela permet notamment d’ajouter progressivement la fonctionnalité des tâches à une application, avant que l’infrastructure nécessaire ne soit mise en place.

Pour l’utiliser, définissez BACKEND à "django.tasks.backends.immediate.ImmediateBackend":

TASKS = {"default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend"}}

Le moteur ImmediateBackend peut aussi être utile pour les tests, afin d’éviter d’avoir à exécuter un processus d’exécution externe au lancement des tests.

Le moteur bidon¶

Le moteur DummyBackend n’exécute pas du tout les tâches en liste d’attente, elle ne fait que stocker leur résultat en vue d’une utilisation ultérieure. Ces résultats de tâches restent toujours dans l’état READY (prête).

Ce moteur n’est pas conçu pour être utilisé en production, il n’est fourni que par commodité et destiné à être utilisé durant le développement et les tests.

Pour l’utiliser, définissez BACKEND à "django.tasks.backends.dummy.DummyBackend":

TASKS = {"default": {"BACKEND": "django.tasks.backends.dummy.DummyBackend"}}

Les résultats des tâches en file d’attente peuvent être obtenus à partir de l’attribut results:

>>> from django.tasks import default_task_backend
>>> my_task.enqueue()
>>> len(default_task_backend.results)
1

Les résultats stockés peuvent être effacés en utilisant la méthode clear():

>>> default_task_backend.clear()
>>> len(default_task_backend.results)
0

Moteurs de tierce partie¶

Comme mentionné au début de cette section, Django fournit uniquement des moteurs adaptés au développement et aux tests. Les systèmes en production doivent se baser sur des moteurs qui fournissent un processus de travail et une implémentation durable de file d’attente. Pour utiliser un moteur de tâches externe avec Django, placez un chemin d’importation Python dans l’option BACKEND du réglage TASKS, comme ceci :

TASKS = {
    "default": {
        "BACKEND": "path.to.backend",
    }
}

Un moteur de tâches est une classe héritant de BaseTaskBackend. Au minimum, elle doit implémenter BaseTaskBackend.enqueue(). Si vous construisez votre propre moteur, vous pouvez utiliser les moteurs de tâches intégrés comme référence d’implémentation. Vous trouverez le code dans le répertoire django/tasks/backends/ du code source de Django.

Gestion du code asynchrone¶

La prise en charge des moteurs de tâches asynchrones est en développement.

BaseTaskBackend possède des variantes asnychrones de toutes les méthodes de base. Par convention, les versions asynchrones de toutes les méthodes sont préfixées par a. Les arguments des deux variantes sont identiques.

Obtention des moteurs¶

Les moteurs peuvent être obtenus en utilisant le gestionnaire de connexion task_backends:

from django.tasks import task_backends

task_backends["default"]  # The default backend
task_backends["reserve"]  # Another backend

Le moteur par défaut est disponible en tant que default_task_backend:

from django.tasks import default_task_backend

Définition des tâches¶

Les tâches sont définies en utilisant le décorateur django.tasks.task() sur une fonction au niveau d’un module :

from django.core.mail import send_mail
from django.tasks import task


@task
def email_users(emails, subject, message):
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

La valeur renvoyée par le décorateur est une instance Task.

Les attributs de Task peuvent être personnalisés via les paramètres du décorateur @task:

from django.core.mail import send_mail
from django.tasks import task


@task(priority=2, queue_name="emails")
def email_users(emails, subject, message):
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

Par convention, les tâches sont définies dans un fichier tasks.py, mais ceci n’est pas obligatoire.

Contexte des tâches¶

Parfois, la tâche à exécuter peut avoir besoin de connaître le contexte dans lequel elle a été mise en file d’attente, et comment elle est exécutée. Elle peut pour cela accepter un argument context qui est une instance de TaskContext.

Pour recevoir le contexte de la tâche comme argument d’une fonction de tâche, passez takes_context au moment de sa définition :

import logging
from django.core.mail import send_mail
from django.tasks import task

logger = logging.getLogger(__name__)


@task(takes_context=True)
def email_users(context, emails, subject, message):
    logger.debug(
        f"Attempt {context.attempt} to send user email. Task result id: {context.task_result.id}."
    )
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

Modification des tâches¶

Avant de mettre une tâche en file d’attente, il peut être nécessaire de modifier certains paramètres de la tâche. Par exemple, pour lui donner une priorité plus haute que la normale.

Une instance Task ne peut pas être modifiée directement. Mais on peut par contre créer une instance modifiée avec la méthode using(), l’originale restant intacte. Par exemple :

>>> email_users.priority
0
>>> email_users.using(priority=10).priority
10

Enqueueing Tasks¶

To add the Task to the queue store, so it will be executed, call the enqueue() method on it. If the Task takes arguments, these can be passed as-is. For example:

result = email_users.enqueue(
    emails=["user@example.com"],
    subject="You have a message",
    message="Hello there!",
)

This returns a TaskResult, which can be used to retrieve the result of the Task once it has finished executing.

To enqueue Tasks in an async context, aenqueue() is available as an async variant of enqueue().

Because both Task arguments and return values are serialized to JSON, they must be JSON-serializable:

>>> process_data.enqueue(datetime.now())
Traceback (most recent call last):
...
TypeError: Object of type datetime is not JSON serializable

Arguments must also be able to round-trip through a json.dumps()/ json.loads() cycle without changing type. For example, consider this Task:

@task()
def double_dictionary(key):
    return {key: key * 2}

With the ImmediateBackend configured as the default backend:

>>> result = double_dictionary.enqueue((1, 2, 3))
>>> result.status
FAILED
>>> result.errors[0].traceback
Traceback (most recent call last):
...
TypeError: unhashable type: 'list'

The double_dictionary Task fails because after the JSON round-trip the tuple (1, 2, 3) becomes the list [1, 2, 3], which cannot be used as a dictionary key.

En général, les objets complexes tels que des instances de modèles ou des types Python comme datetime ou tuple ne peuvent pas être utilisés dans le système des tâches sans une conversion spécifique.

Transactions¶

Pour la plupart des moteurs, les tâches sont exécutées dans un processus séparé, utilisant une connexion de base de données différente. Lorsqu’on utilise une transaction, sans attendre que celle-ci soit commitée, des processus d’exécution pourraient démarrer le traitement d’une tâche utilisant des objets auxquels elle ne peut pas encore accéder.

Par exemple, considérons cet exemple simplifié :

@task
def my_task(thing_num):
    Thing.objects.get(num=thing_num)


with transaction.atomic():
    Thing.objects.create(num=1)
    my_task.enqueue(thing_num=1)

Pour éviter le scénario où my_task s’exécute avant que Thing ne soit commité dans la base de données, utilisez transaction.on_commit(), en liant tous les arguments à destination de enqueue() via functools.partial():

from functools import partial

from django.db import transaction

with transaction.atomic():
    Thing.objects.create(num=1)
    transaction.on_commit(partial(my_task.enqueue, thing_num=1))

Task results¶

When enqueueing a Task, you receive a TaskResult, however it’s likely useful to retrieve the result from somewhere else (for example another request or another Task).

Each TaskResult has a unique id, which can be used to identify and retrieve the result once the code which enqueued the Task has finished.

The get_result() method can retrieve a result based on its id:

# Later, somewhere else...
result = email_users.get_result(result_id)

To retrieve a TaskResult, regardless of which kind of Task it was from, use the get_result() method on the backend:

from django.tasks import default_task_backend

result = default_task_backend.get_result(result_id)

To retrieve results in an async context, aget_result() is available as an async variant of get_result() on both the backend and Task.

Certains moteurs, tels que le moteur intégré ImmediateBackend, ne prennent pas en charge get_result(). Si on appelle cette méthode avec ces moteurs, une erreur NotImplementedError sera générée.

Mise à jour des résultats¶

Un objet TaskResult contient l’état d’exécution d’une tâche au moment où il est obtenu. Si la tâche se termine après que get_result() est appelée, l’objet ne se met pas à jour tout seul.

Pour actualiser les valeurs, appelez la méthode django.tasks.TaskResult.refresh():

>>> result.status
RUNNING
>>> result.refresh()  # or await result.arefresh()
>>> result.status
SUCCESSFUL

Return values¶

If your Task function returns something, it can be retrieved from the django.tasks.TaskResult.return_value attribute:

>>> result.status
SUCCESSFUL
>>> result.return_value
42

If the Task has not finished executing, or has failed, ValueError is raised.

>>> result.status
RUNNING
>>> result.return_value
Traceback (most recent call last):
...
ValueError: Task has not finished yet

Errors¶

If the Task doesn’t succeed, and instead raises an exception, either as part of the Task or as part of running it, the exception and traceback are saved to the django.tasks.TaskResult.errors list.

Each entry in errors is a TaskError containing information about error raised during the execution:

>>> result.errors[0].exception_class
<class 'ValueError'>

Note that this is just the type of exception, and contains no other values. The traceback information is reduced to a string which you can use to help debugging:

>>> result.errors[0].traceback
Traceback (most recent call last):
...
TypeError: Object of type datetime is not JSON serializable