Skip to content

Django Celery Task Track Fix

DodaTech Updated 2026-06-24 2 min read

In this tutorial, you'll learn about Django Celery Task Track Fix. We cover key concepts, practical examples, and best practices.

The Problem

You launch a Celery task but have no idea if it succeeded, failed, or is still running. Without result tracking, debugging async failures is difficult.

Quick Fix

Wrong — no result backend

# settings.py
CELERY_RESULT_BACKEND = None

# views.py
def start_task(request):
    my_task.delay()  # Task runs, but we never know the result
    return Response({'status': 'started'})

Output: Task starts but status, result, and traceback are invisible.

Correct — Redis result backend

# settings.py
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_EXTENDED = True

# tasks.py
@app.task(bind=True)
def long_task(self, arg):
    self.update_state(state='PROGRESS', meta={'progress': 0})
    # ... do work ...
    self.update_state(state='PROGRESS', meta={'progress': 50})
    # ... do work ...
    return {'result': 'done', 'progress': 100}

Output: Task status is stored in Redis and queryable by task ID.

Checking task status

from celery.result import AsyncResult

def task_status(request, task_id):
    result = AsyncResult(task_id)
    return Response({
        'task_id': task_id,
        'state': result.state,
        'status': result.status,
        'result': result.result,
        'traceback': result.traceback,
    })

Polling from frontend

# views.py
def start_and_poll(request):
    task = my_task.delay()
    return Response({
        'task_id': task.id,
        'status_url': f'/tasks/{task.id}/status/',
    })

# JavaScript (client-side polling)
"""
const resp = await fetch('/api/start-task/');
const { task_id, status_url } = await resp.json();
const interval = setInterval(async () => {
    const status = await fetch(status_url);
    const data = await status.json();
    if (data.state === 'SUCCESS' || data.state === 'FAILURE') {
        clearInterval(interval);
    }
}, 2000);
"""

Task success/failure callbacks

@app.task(bind=True, on_success=handle_success, on_failure=handle_failure)
def my_task(self):
    ...

def handle_success(self, retval, task_id, args, kwargs):
    notify_success(task_id, retval)

def handle_failure(self, exc, task_id, args, kwargs, einfo):
    notify_failure(task_id, str(exc))

Prevention

  • Always set CELERY_RESULT_BACKEND in production.
  • Store task IDs and expose status endpoints for long-running tasks.
  • Use CELERY_TASK_TRACK_STARTED = True to see tasks currently running.

Common Mistakes with celery task track

  1. Non-exhaustive pattern matches that compile with warnings then crash at runtime
  2. Misunderstanding that String is [Char] with poor performance for large text operations
  3. Using foldl instead of foldl' causing stack overflow on large lists

These mistakes appear frequently in real-world DJANGO code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.

Practice Exercise

Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.

This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.

FAQ

### What result backends are supported?

Redis, RabbitMQ (rpc), database (Django ORM), S3, Cassandra, Elasticsearch.

How long are task results stored?

Default is 24 hours (CELERY_RESULT_EXPIRES). Set it as needed: CELERY_RESULT_EXPIRES = 3600 (1 hour).

Can I store results in the database?

Yes: CELERY_RESULT_BACKEND = 'django-db' with django-celery-results installed.

← Previous Django Celery Group Result Fix Next → Django Channels Group Send Fix

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Quick Fix