Django Tasks

0.12.0 · active · verified Thu Apr 09

A backport of Django's built-in Tasks framework, `django-tasks` provides a robust way to run asynchronous background tasks within Django applications. As of version 0.12.0, it focuses on mirroring the upstream `django.tasks` package, with backends like DB and RQ now available as separate packages. It maintains an active release cadence, frequently updating to support new Django and Python versions.

Warnings

breaking In version `0.12.0`, the Database and RQ backends were extracted into separate packages (`django-tasks-db` and `django-tasks-rq`). If you previously used these backends, `django-tasks` no longer includes them directly.
Fix: Install the new packages (e.g., `pip install django-tasks-db django-tasks-rq`) and update your `TASKS_BACKEND` setting in `settings.py` to point to the new backend paths (e.g., `'tasks_db.backends.DatabaseBackend'` or `'tasks_rq.backends.RQBackend'`).
breaking The `enqueue_on_commit` functionality was removed in version `0.10.0`.
Fix: Tasks should be enqueued directly using `.delay()`. If deferred execution within a transaction is required, you will need to implement custom transaction management logic or use a different mechanism.
breaking In version `0.8.0`, the `TaskResult` status `NEW` was renamed to `READY` to better support retry functionality. Additionally, exceptions and tracebacks are now stored in a `.errors` list on the `TaskResult` model, rather than directly on `exception` and `traceback` attributes.
Fix: Update any code referencing `TaskResult.status == 'NEW'` to `TaskResult.status == 'READY'`. Access exceptions and tracebacks via the `task_result.errors` list (e.g., `task_result.errors[0].exception_class`, `task_result.errors[0].traceback`).
breaking In version `0.6.0`, the `TaskResult` status `completed` was renamed to `succeeded`.
Fix: Update any code referencing `TaskResult.status == 'completed'` to `TaskResult.status == 'succeeded'`.

Install

pip install django-tasks Install core library

Imports

task
```
from tasks import task
```
The decorator is imported from the top-level 'tasks' module, not 'django_tasks'.
TaskResult
```
from tasks.models import TaskResult
```
Models are accessed via the 'tasks.models' path.
run_task
```
from tasks import run_task
```
Used for manual synchronous task execution.

Quickstart

This quickstart demonstrates how to define a task using the `@task` decorator and enqueue it for asynchronous execution using `.delay()`. It includes a minimal Django configuration for testing, assuming the `django-tasks-db` backend is used and installed.

import os
from django.conf import settings

# Minimal Django setup for demonstration
if not settings.configured:
    settings.configure(
        INSTALLED_APPS=[
            'django.contrib.auth',
            'django.contrib.contenttypes',
            'tasks' # Ensure 'tasks' is in INSTALLED_APPS
        ],
        DATABASES={
            'default': {
                'ENGINE': 'django.db.backends.sqlite3',
                'NAME': ':memory:',
            }
        },
        TASKS_BACKEND='tasks_db.backends.DatabaseBackend', # Or 'tasks_rq.backends.RQBackend'
        USE_TZ=True
    )

# Ensure migrations are run for tasks_db if using it
# In a real project, this would be part of `manage.py migrate`
import django
django.setup()

# This part assumes django-tasks-db is installed for the example
# and that tasks.backends is configured.
# For a true quickstart, you'd run `python manage.py makemigrations tasks_db && python manage.py migrate`

from tasks import task

@task
def greet_user(name):
    print(f"Hello, {name}!")
    return f"Greeting for {name} completed."

# Enqueue the task for asynchronous execution
result = greet_user.delay("World")
print(f"Task enqueued with ID: {result.pk}")

# To process the task (typically run by a worker process)
# For local testing, you might manually run the task
# from tasks import run_task
# run_task(result.pk)

# You can inspect the result object (needs refresh if worker is separate)
# result.refresh_from_db()
# print(f"Task status: {result.status}")
# print(f"Task result: {result.result}")

view raw JSON →