Exécuter des tâches en arrière-plan dans Django sans Celery

Pourquoi utiliser une tâche en arrière-plan ?

Dans la partie 6, nous avons créé le pipeline qui transforme un CV en questions d'entretien : analyse, validation, structuration, génération et enregistrement. Il reste toutefois un problème : c'est lent. Analyser un fichier puis effectuer plusieurs appels IA séquentiels peut prendre plusieurs secondes. Vous ne pouvez donc pas demander au navigateur d'un utilisateur d'attendre aussi longtemps sur une seule requête HTTP. Nous devons exécuter ces tâches Django en arrière-plan, en dehors du cycle de la requête.

La réponse habituelle s'appelle Celery. Mais Celery implique de faire fonctionner un broker comme Redis, des processus supplémentaires et une véritable infrastructure : beaucoup de contraintes pour un MVP. Cette application adopte une approche plus légère, sans service supplémentaire : elle lance un process worker distinct pour chaque tâche, puis laisse le frontend interroger régulièrement l'avancement. Voici comment tout cela s'articule.

Un registre de tâches minimal

Commençons par un moyen d'enregistrer les tâches en arrière-plan par leur nom. Il s'agit d'un décorateur très simple qui stocke les fonctions dans un dictionnaire, afin qu'une tâche puisse ensuite être retrouvée à partir d'une chaîne de caractères :

TASK_REGISTRY = {}


def register_task(name):
    def decorator(func):
        TASK_REGISTRY[name] = func
        return func
    return decorator


def get_task(name):
    return TASK_REGISTRY.get(name)

La fonction process_interview de la partie 6 reçoit alors simplement un décorateur et devient accessible par son nom :

@register_task("process_interview")
def process_interview(task, payload):
    interview = Interview.objects.get(id=payload["interview_id"])
    # ... parse, validate, generate questions (from part 6)

Pourquoi utiliser un registre plutôt que d'appeler directement la fonction ? Parce que le worker qui exécute la tâche se trouve dans un process distinct et ne reçoit qu'une chaîne ("process_interview") : il lui faut un moyen de convertir ce nom en fonction. C'est précisément le rôle du registre. Il permet aussi de retrouver facilement toutes les tâches en arrière-plan au même endroit.

Lancer la tâche

Lorsqu'un utilisateur envoie un CV, la vue effectue uniquement le nécessaire de façon synchrone : créer l'entretien, créer l'enregistrement de la tâche, lancer le worker, puis répondre immédiatement avec un 202 Accepted. Le client comprend ainsi que la demande a bien été reçue et qu'il doit vérifier l'avancement :

interview = Interview.objects.create(
    candidate_name=full_name,
    candidate_email=email,
    cv_file=resume_file,
    device=request.device,
    client_ip=client_ip,
)

task = BackgroundTask.objects.create(
    type="process_interview",
    payload={"interview_id": interview.id},
    name=f"Process Interview {interview.interview_uuid}",
)
start_background_task(task.id)

return Response({"data": {"id": str(interview.interview_uuid)}}, status=202)

Le détail important est le code 202, et non 200. C'est le code HTTP honnête pour indiquer que la demande a été acceptée pour traitement, mais qu'elle n'est pas encore terminée. Le travail lourd n'a pas encore été exécuté : la réponse revient en quelques millisecondes et le pipeline s'exécute ailleurs.

Lancer un processus worker

Voici l'élément qui permet de fonctionner sans Celery. start_background_task lance un processus complètement distinct avec le module Python subprocess, lequel exécute une commande de gestion Django :

import subprocess
import sys

from core.models import BackgroundTask


def start_background_task(task_id):
    task = BackgroundTask.objects.get(id=task_id)
    if task.status in [2, 3]:  # already running or done
        task.append_log("Task already running or completed", True)
        return task

    subprocess.Popen(
        [sys.executable, "manage.py", "background_worker", f"--task_id={task.id}"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    return task

Deux choix sont délibérés ici. start_new_session=True détache le worker du processus web : il continue donc à s'exécuter seul, même lorsque la réponse HTTP a été envoyée depuis longtemps. La sortie est dirigée vers DEVNULL, car ce processus fonctionne en mode « fire-and-forget » : ses véritables journaux sont les champs de statut et de log qu'il écrit dans la base de données, et que nous pourrons ensuite interroger.

Pourquoi utiliser un processus distinct plutôt qu'un thread ? Pour l'isolation. Un thread partage la mémoire et le processus du serveur web : une tâche qui plante sévèrement ou consomme trop de mémoire peut entraîner tout le serveur avec elle. Un subprocess est cloisonné : s'il s'arrête, votre application web ne le remarque même pas. Cette isolation est particulièrement précieuse lorsque le traitement implique de gros fichiers et des appels IA imprévisibles.

La commande du worker

Le processus lancé exécute une commande de gestion dont le seul rôle est de retrouver la tâche, de l'exécuter et d'enregistrer le résultat :

import os

from django.core.management.base import BaseCommand

from core.models import BackgroundTask
from core.registry import get_task
from tasks import *  # noqa — importing runs the @register_task decorators


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("--task_id", type=str)

    def handle(self, *args, **options):
        task = BackgroundTask.objects.get(id=options["task_id"])
        task.status = 2  # running
        task.pid = os.getpid()
        task.save(update_fields=["status", "pid", "updated_at"])

        func = get_task(task.type)
        if not func:
            task.append_log("Task type not registered", commit=True)
            return

        try:
            func(task, task.payload)
            task.status = 3  # done
        except Exception as e:
            task.status = 4  # error
            task.append_log(str(e))
        task.save(update_fields=["status", "log", "updated_at"])

Remarquez la ligne from tasks import * : cet import exécute réellement les décorateurs @register_task et remplit le registre. Ainsi, get_task peut retrouver la fonction. La suite est simple : marquer la tâche comme en cours, l'exécuter, puis la marquer comme terminée ou en erreur. Chaque changement d'état est écrit dans la base de données, ce qui permet précisément au frontend de suivre la progression depuis l'extérieur.

Interroger le statut

Puisque le traitement s'effectue indépendamment de la requête, le frontend doit pouvoir demander « est-ce terminé ? ». Il suffit d'un endpoint qui lit le statut actuel de l'entretien :

class InterviewStatusAPIView(APIView):
    INTERVIEW_STATUS = {
        1: "Pending", 2: "Processing Resume", 3: "Preparing Interview",
        4: "In progress", 7: "Completed", 8: "Failed",
    }

    def get(self, request, *args, **kwargs):
        interview = Interview.objects.get(
            interview_uuid=kwargs["interview_id"],
            device=request.device,
        )
        return Response({
            "status": interview.status,
            "status_text": self.INTERVIEW_STATUS.get(interview.status),
        })

Le frontend appelle cet endpoint toutes les deux secondes environ et actualise l'interface à mesure que le statut passe de « Processing Resume » à « Preparing Interview », puis à l'état prêt. Il s'agit d'un short polling : ce n'est pas l'option la plus élégante en théorie, mais pour une tâche ponctuelle comme celle-ci, c'est simple, fiable et cela ne nécessite ni websockets ni infrastructure supplémentaire. Pour un MVP, choisir volontairement l'option la plus simple, sans dépendance, est souvent la bonne décision.

Ce qu'il faudra renforcer en production

Vous disposez maintenant du mécanisme complet : un registre, un worker subprocess lancé en mode fire-and-forget et un suivi par interrogation du statut. C'est suffisant pour exécuter tout le pipeline d'entretien en dehors du cycle de la requête, et pour finaliser une application réellement fonctionnelle. C'est une étape importante : à ce stade, l'entretien textuel fonctionne de bout en bout.

Ce que cette approche légère ne fournit pas, mais qu'un véritable déploiement finira par exiger, c'est la couche opérationnelle la plus complexe : limiter la mémoire qu'un worker incontrôlable peut utiliser, arrêter ou tuer une tâche bloquée, réessayer en cas d'échec et terminer automatiquement les entretiens qui restent sans réponse. Ce sont précisément les fonctionnalités que Celery fournit directement. En renonçant à Celery, vous devez donc construire vous-même celles dont vous avez réellement besoin. La version renforcée de ce système de tâches — avec limites mémoire, arrêt sécurisé et reprise après incident — fait partie du code source complet inclus dans le kit de démarrage. Sinon, passez à la partie 8, où l'application se met enfin à parler : nous ajouterons la saisie vocale pour que les candidats répondent à voix haute, comme dans un véritable entretien.