Running Background Tasks in Django Without Celery

Why This Needs a Background Task

In part 6 we built the pipeline that turns a CV into interview questions: parse, validate, structure, generate, save. There's just one problem — it's slow. Parsing a file plus several sequential AI calls can take many seconds, and you cannot make a user's browser sit and wait on a single HTTP request for that long. So we need to run background tasks in Django, off the request cycle.

The usual answer is Celery. But Celery means running a broker like Redis, extra processes, and real infrastructure — a lot to take on for an MVP. This app takes a lighter path that needs no extra services: spawn a separate worker process for each job, and let the frontend poll for progress. Here's how it fits together.

A Tiny Task Registry

First, a way to register background jobs by name. This is a dead-simple decorator that stores functions in a dictionary, so a task can be looked up later by a string:

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)

Now the process_interview function from part 6 just gets a decorator, and it becomes discoverable by name:

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

Why bother with a registry instead of calling the function directly? Because the worker that runs the job lives in a separate process and only receives a string ("process_interview") — it needs a way to turn that name back into a function. The registry is that lookup. It also keeps every background job discoverable in one place.

Kicking Off the Task

When a user uploads a CV, the view does the minimum synchronously — create the interview, create a task record, launch the worker — then returns immediately with a 202 Accepted, telling the client "we've got it, keep checking back":

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)

The key detail is 202, not 200. It's the honest HTTP status for "accepted for processing, but not finished." The heavy work hasn't run yet — the response comes back in milliseconds, and the actual pipeline runs elsewhere.

Spawning a Worker Process

Here's the piece that makes it work without Celery. start_background_task spawns a completely separate process with Python's subprocess module, that runs a Django management command:

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

Two choices here are deliberate. start_new_session=True detaches the worker from the web process, so it keeps running on its own even after the HTTP response is long gone. And output goes to DEVNULL because this process is fire-and-forget — its real "logging" is the status and log fields it writes to the database, which is what we can actually query later.

Why a separate process instead of a thread? Isolation. A thread shares the web server's memory and process — a task that crashes hard or eats too much memory can take the whole server down with it. A subprocess is walled off: if it dies, your web app doesn't even notice. That isolation is worth a lot when the work involves large files and unpredictable AI calls.

The Worker Command

The spawned process runs a management command whose only job is to look up the task, run it, and record what happened:

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"])

Notice the from tasks import * line — that import is what actually runs your @register_task decorators and populates the registry, so get_task can find the function. From there it's straightforward: mark the task running, run it, and mark it done or error. Every state change is written to the database, which is exactly what lets the frontend see progress from the outside.

Polling for Status

Since the work happens out of band, the frontend needs a way to ask "are we there yet?" That's a plain endpoint that reads the interview's current status:

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),
        })

The frontend calls this every couple of seconds and updates the UI as the status climbs from "Processing Resume" to "Preparing Interview" to ready. This is short polling — not the most elegant option in the abstract, but for an occasional, one-off job like this it's simple, reliable, and needs no websockets or extra infrastructure. Choosing the boring, dependency-free option on purpose is often the right call for an MVP.

Where the Production Version Goes Further

You now have the full mechanism: a registry, a fire-and-forget subprocess worker, and status polling — enough to run the whole interview pipeline off the request cycle, and enough to finish a real, working app. That's the milestone: by this point the text-based interview runs end to end.

What this lightweight approach doesn't give you, that a real deployment eventually wants, is the harder operational layer: capping how much memory a runaway worker can use, stopping or killing a stuck task, retrying on failure, and auto-completing interviews that stall. Those are exactly the things Celery gives you out of the box, and the trade-off for skipping Celery is that you build the few you actually need yourself. The production-hardened version of this task system — memory limits, safe task stopping, and recovery — is part of the finished source in the starter kit. Otherwise, part 8 is next, where the app finally speaks: adding voice input so candidates answer out loud, the way a real interview works.