Tasks Framework

# config/settings.py (excerpt)
import os

from django.core.management.utils import get_random_secret_key

SECRET_KEY = os.environ.get("DJANGO_SECRET_KEY") or get_random_secret_key()

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "catalog",
]

MAILERS = {"default": {"BACKEND": "django.core.mail.backends.console.EmailBackend"}}

# Django's default: run every task immediately, in the calling thread.
TASKS = {
    "default": {
        "BACKEND": "django.tasks.backends.immediate.ImmediateBackend",
        "QUEUES": ["default", "emails"],
    }
}

# catalog/models.py
from django.conf import settings
from django.db import models


class Book(models.Model):
    title = models.CharField(max_length=200)
    isbn = models.CharField(max_length=13, unique=True)

    def __str__(self):
        return self.title


class Review(models.Model):
    class Status(models.TextChoices):
        PENDING = "pending", "Pending"
        APPROVED = "approved", "Approved"

    book = models.ForeignKey(Book, on_delete=models.CASCADE, related_name="reviews")
    reviewer = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="reviews"
    )
    rating = models.PositiveSmallIntegerField()
    body = models.TextField(blank=True)
    status = models.CharField(max_length=10, choices=Status, default=Status.PENDING)
    moderators_notified_at = models.DateTimeField(null=True, blank=True)

    def __str__(self):
        return f"{self.reviewer} on {self.book}: {self.rating}/5"

# catalog/tasks.py
import logging

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

from .models import Review

logger = logging.getLogger(__name__)


@task
def add(a, b):
    return a + b


@task(queue_name="emails", takes_context=True)
def notify_moderators(context, review_id):
    """Email the moderators about a new review; safe to run twice."""
    try:
        review = Review.objects.select_related("book").get(pk=review_id)
    except Review.DoesNotExist:
        logger.warning("Review %s is gone, nothing to do", review_id)
        return "gone"
    if review.moderators_notified_at is not None:  # idempotency guard
        return "already-notified"
    logger.info("attempt %s for result %s", context.attempt, context.task_result.id)
    send_mail(
        subject=f"New review for {review.book.title}",
        message=f"Rating {review.rating}/5 is waiting for moderation.",
        from_email=None,
        recipient_list=["moderators@example.com"],
    )
    Review.objects.filter(pk=review.pk).update(moderators_notified_at=timezone.now())
    return "sent"

from django.tasks import task


def outer():
    @task
    def inner():  # InvalidTask: not defined at module level
        pass


@task(takes_context=True)
def bad(x):  # InvalidTask: first argument must be named "context"
    pass


@task(run_after=None)  # TypeError: run_after is set per call with using()
def bad2():
    pass


@task(backend="nope")  # InvalidTaskBackend: no such alias in TASKS
def bad3():
    pass

>>> from catalog.tasks import add
>>> result = add.enqueue(2, 3)
>>> result.status
<TaskResultStatus.SUCCESSFUL: 'SUCCESSFUL'>
>>> result.return_value
5
>>> result.attempts
1
>>> result.is_finished
True
>>> add.priority, add.queue_name, add.backend, add.run_after, add.name
(0, 'default', 'default', None, 'add')

>>> from datetime import timedelta
>>> from django.utils import timezone
>>> add.using(priority=10).priority
10
>>> add.priority
0
>>> add.using(queue_name="nope").enqueue(1, 2)
InvalidTask: Queue 'nope' is not valid for backend.
>>> add.using(priority=101).enqueue(1, 2)
InvalidTask: priority must be a whole number between -100 and 100.
>>> add.using(run_after=timezone.now() + timedelta(minutes=1)).enqueue(1, 2)
InvalidTask: Backend does not support run_after.

INFO attempt 1 for result <result id>

>>> from django.tasks import task
>>> @task
... def explode(label):
...     raise ValueError(f"cannot process {label}")
>>> result = explode.enqueue("x")
>>> result.status
<TaskResultStatus.FAILED: 'FAILED'>
>>> result.errors[0].exception_class
<class 'ValueError'>
>>> result.errors[0].traceback.splitlines()[-1]
'ValueError: cannot process x'
>>> result.return_value
Traceback (most recent call last):
...
ValueError: Task failed

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

>>> from django.tasks import default_task_backend, task_backends
>>> task_backends["default"]
<django.tasks.backends.immediate.ImmediateBackend object at 0x...>
>>> default_task_backend.queues, default_task_backend.options, default_task_backend.alias
({'default', 'emails'}, {}, 'default')
>>> default_task_backend.supports_defer, default_task_backend.supports_priority
(False, True)
>>> [b.alias for b in task_backends.all()]
['default', 'json']

# catalog/services.py
from functools import partial

from django.db import transaction

from .models import Review
from .tasks import notify_moderators


def create_review(book, reviewer, rating, body=""):
    """Write the review, and enqueue the follow-up only if the write commits."""
    with transaction.atomic():
        review = Review.objects.create(
            book=book, reviewer=reviewer, rating=rating, body=body
        )
        transaction.on_commit(partial(notify_moderators.enqueue, review_id=review.pk))
    return review

# catalog/views.py (excerpt)
import json

from django.http import JsonResponse
from django.shortcuts import get_object_or_404
from django.views.decorators.http import require_POST

from .models import Book
from .services import create_review


@require_POST
def submit_review(request, book_id):
    book = get_object_or_404(Book, pk=book_id)
    data = json.loads(request.body)
    review = create_review(book, _reviewer(), data["rating"], data.get("body", ""))
    return JsonResponse({"review": review.pk}, status=201)

$ curl -s -c cookies.txt http://127.0.0.1:47302/catalog/csrf/
{"csrf": true}
$ TOKEN=$(awk '$6 == "csrftoken" {print $7}' cookies.txt)
$ curl -s -b cookies.txt -H "X-CSRFToken: $TOKEN" -H 'Content-Type: application/json' \
    -d '{"rating": 4}' -w ' [%{http_code}]\n' http://127.0.0.1:47302/catalog/books/1/reviews/
{"review": 1} [201]
$ curl -s -b cookies.txt -H 'Content-Type: application/json' \
    -d '{"rating": 4}' -o /dev/null -w 'no token: %{http_code}\n' http://127.0.0.1:47302/catalog/books/1/reviews/
no token: 403

# catalog/views.py (excerpt)
from django.contrib.auth import get_user_model
from django.http import JsonResponse
from django.views.decorators.http import require_POST

from .models import Book, Review
from .tasks import notify_moderators


@require_POST
async def submit_review_async(request, book_id):
    book = await Book.objects.aget(pk=book_id)
    data = json.loads(request.body)
    reviewer, _ = await get_user_model().objects.aget_or_create(username="reader")
    review = await Review.objects.acreate(
        book=book, reviewer=reviewer, rating=data["rating"], body=data.get("body", "")
    )
    result = await notify_moderators.aenqueue(review_id=review.pk)
    return JsonResponse(
        {"review": review.pk, "task_result": result.id, "status": result.status},
        status=202,
    )

$ curl -s -b cookies.txt -H "X-CSRFToken: $TOKEN" -H 'Content-Type: application/json' \
    -d '{"rating": 4}' -w ' [%{http_code}]\n' http://127.0.0.1:47302/catalog/books/1/reviews/async/
{"review": 2, "task_result": "<result id>", "status": "SUCCESSFUL"} [202]

# catalog/backends.py
import json
from dataclasses import dataclass
from pathlib import Path
from uuid import uuid4

from django.tasks import Task, TaskResult, TaskResultStatus
from django.tasks.backends.base import BaseTaskBackend
from django.utils import timezone


@dataclass(frozen=True, slots=True, kw_only=True)
class RetryingTask(Task):
    """A Task that also carries backend-specific data (Django 6.1: task(**kwargs))."""

    max_attempts: int = 1


class JsonLinesBackend(BaseTaskBackend):
    """Teaching backend: appends each enqueued task to a JSON Lines file."""

    task_class = RetryingTask
    supports_priority = False

    def __init__(self, alias, params):
        super().__init__(alias, params)
        self.path = Path(self.options["path"])

    def enqueue(self, task, args, kwargs):
        self.validate_task(task)
        result = TaskResult(
            task=task,
            id=uuid4().hex,
            status=TaskResultStatus.READY,
            enqueued_at=timezone.now(),
            started_at=None,
            finished_at=None,
            last_attempted_at=None,
            args=args,
            kwargs=kwargs,
            backend=self.alias,
            errors=[],
            worker_ids=[],
        )
        line = {
            "id": result.id,
            "task": task.module_path,
            "queue": task.queue_name,
            "max_attempts": task.max_attempts,
            "args": result.args,
            "kwargs": result.kwargs,
        }
        with self.path.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(line) + "\n")
        return result

# config/settings.py (add the alias) and catalog/tasks.py (use it)
TASKS = {
    "default": {...},  # as before
    "json": {
        "BACKEND": "catalog.backends.JsonLinesBackend",
        "OPTIONS": {"path": BASE_DIR / "tasks.jsonl"},
    },
}

@task(backend="json", max_attempts=3)
def archive_review(review_id):
    return f"archived {review_id}"

# catalog/management/commands/run_json_tasks.py
import json

from django.core.management.base import BaseCommand
from django.tasks import task_backends
from django.utils.module_loading import import_string


class Command(BaseCommand):
    help = "Run every task recorded by the JsonLinesBackend (a teaching worker)."

    def add_arguments(self, parser):
        parser.add_argument("--backend", default="json")

    def handle(self, *args, backend, **options):
        path = task_backends[backend].path
        for line in path.read_text(encoding="utf-8").splitlines():
            job = json.loads(line)
            task = import_string(job["task"])  # a Task instance
            for attempt in range(1, job["max_attempts"] + 1):
                try:
                    value = task.call(*job["args"], **job["kwargs"])
                except Exception as exc:
                    self.stderr.write(f"{job['id'][:8]} attempt {attempt} failed: {exc}")
                else:
                    self.stdout.write(f"{job['id'][:8]} {task.name} -> {value!r}")
                    break
        path.write_text("", encoding="utf-8")

>>> from catalog.tasks import archive_review
>>> type(archive_review).__name__, archive_review.max_attempts, archive_review.backend
('RetryingTask', 3, 'json')
>>> archive_review.enqueue(7).status
<TaskResultStatus.READY: 'READY'>
>>> archive_review.using(priority=5).enqueue(8)
InvalidTask: Backend does not support setting priority of tasks.

$ python manage.py run_json_tasks
26a77163 archive_review -> 'archived 7'

# catalog/tests.py (excerpt)
import json

from django.contrib.auth import get_user_model
from django.core import mail
from django.db import transaction
from django.tasks import default_task_backend, TaskResultStatus
from django.tasks.exceptions import InvalidTask, TaskResultDoesNotExist, TaskResultMismatch
from django.test import TestCase, override_settings
from django.urls import reverse

from .models import Book, Review
from .tasks import add, notify_moderators

DUMMY = {"default": {"BACKEND": "django.tasks.backends.dummy.DummyBackend", "QUEUES": ["default", "emails"]}}
IMMEDIATE = {"default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend", "QUEUES": ["default", "emails"]}}


@override_settings(TASKS=DUMMY)
class DummyTests(TestCase):
    def setUp(self):
        default_task_backend.clear()
        self.book = Book.objects.create(title="Don Quijote", isbn="9780142437230")
        self.user = get_user_model().objects.create(username="reader")

    def test_results_are_stored_not_run(self):
        result = add.enqueue(2, 3)
        self.assertEqual(result.status, TaskResultStatus.READY)
        self.assertEqual(default_task_backend.results[0].args, [2, 3])
        with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
            result.return_value

    def test_view_enqueues_after_commit(self):
        url = reverse("catalog:submit-review", args=[self.book.pk])
        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            response = self.client.post(url, json.dumps({"rating": 5}), content_type="application/json")
        self.assertEqual(response.status_code, 201)
        self.assertEqual(len(callbacks), 1)
        stored = default_task_backend.results[0]
        self.assertEqual(stored.task.name, "notify_moderators")
        self.assertEqual(stored.kwargs, {"review_id": response.json()["review"]})
        self.assertEqual(stored.task.queue_name, "emails")

    def test_rollback_means_no_task(self):
        try:
            with transaction.atomic():
                review = Review.objects.create(book=self.book, reviewer=self.user, rating=1)
                transaction.on_commit(lambda: notify_moderators.enqueue(review_id=review.pk))
                raise RuntimeError("boom")
        except RuntimeError:
            pass
        self.assertEqual(len(default_task_backend.results), 0)


@override_settings(TASKS=IMMEDIATE)
class NotifyModeratorsTests(TestCase):
    def setUp(self):
        self.book = Book.objects.create(title="La Galatea", isbn="9780000000011")
        self.user = get_user_model().objects.create(username="reader")
        self.review = Review.objects.create(book=self.book, reviewer=self.user, rating=5)

    def test_sends_once(self):
        first = notify_moderators.enqueue(review_id=self.review.pk)
        second = notify_moderators.enqueue(review_id=self.review.pk)
        self.assertEqual((first.return_value, second.return_value), ("sent", "already-notified"))
        self.assertEqual(len(mail.outbox), 1)
        self.assertEqual(mail.outbox[0].subject, "New review for La Galatea")

    def test_missing_review(self):
        self.assertEqual(notify_moderators.enqueue(review_id=0).return_value, "gone")

from django.core.management import call_command

from .tasks import shout

DATABASE = {"default": {"BACKEND": "django_tasks_db.DatabaseBackend", "QUEUES": ["default", "emails"]}}


@override_settings(TASKS=DATABASE)
class DatabaseBackendTests(TestCase):
    def test_worker_batch_runs_tasks_in_priority_order(self):
        low = shout.using(priority=-5).enqueue("low")
        high = shout.using(priority=50).enqueue("high")
        self.assertEqual(low.status, TaskResultStatus.READY)
        call_command("db_worker", "--batch", "--queue-name", "*", "--no-reload", "--no-startup-delay", verbosity=0)
        low.refresh()
        high.refresh()
        self.assertEqual((low.return_value, high.return_value), ("LOW", "HIGH"))
        self.assertLess(high.started_at, low.started_at)

def test_pickle_round_trip(self):  # new in 6.1
    result = add.enqueue(1, 2)
    self.assertEqual(pickle.loads(pickle.dumps(add)), add)
    self.assertEqual(pickle.loads(pickle.dumps(result)).return_value, 3)