Sample code for the article on Django Tasks

Author: viv1

django-tasks/README.md

@@ -0,0 +1,62 @@
+# Django Tasks: Exploring the Built-in Task Framework in Django 6
+
+Example project for the Real Python tutorial [Django Tasks: Exploring the Built-in Task Framework in Django 6](https://realpython.com/django-tasks/).
+
+The project demonstrates Django 6's `@task` decorator, the `django-tasks-db` database-backed worker, named queues with priorities, and the `transaction.on_commit` pattern for safe enqueueing inside an atomic block.
+
+## Setup
+
+Set up a virtual environment and install dependencies:
+
+```sh
+$ python -m venv venv
+$ source venv/bin/activate
+(venv) $ python -m pip install -r requirements.txt
+```
+
+Run migrations:
+
+```sh
+(venv) $ python -m manage makemigrations myapp
+(venv) $ python -m manage migrate
+```
+
+## Run
+
+The project uses two processes: the Django dev server and the `db_worker`.
+
+In one terminal, start the worker:
+
+```sh
+(venv) $ python -m manage db_worker --queue-name '*'
+```
+
+In a second terminal, start the dev server:
+
+```sh
+(venv) $ python -m manage runserver
+```
+
+Trigger the demo endpoints with `curl` or your browser:
+
+```sh
+$ curl 'http://localhost:8000/register/?email=alice@example.com&name=Alice'
+$ curl http://localhost:8000/checkout/
+$ curl http://localhost:8000/task-status/<task-id>/
+```
+
+You can also enqueue tasks from the Django shell:
+
+```sh
+(venv) $ python -m manage shell
+```
+
+```pycon
+>>> from myapp.tasks import say_hello
+>>> result = say_hello.enqueue("Real Python")
+>>> result.refresh()
+>>> result.status
+SUCCESSFUL
+>>> result.return_value
+'Hello, Real Python!'
+```

django-tasks/config/__init__.py

@@ -0,0 +1,39 @@
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+
+SECRET_KEY = "django-insecure-demo-key-do-not-use-in-production"
+DEBUG = True
+ALLOWED_HOSTS = ["*"]
+
+INSTALLED_APPS = [
+    "django.contrib.contenttypes",
+    "django.contrib.auth",
+    "django_tasks_db",
+    "myapp",
+]
+
+MIDDLEWARE = []
+
+ROOT_URLCONF = "config.urls"
+
+DATABASES = {
+    "default": {
+        "ENGINE": "django.db.backends.sqlite3",
+        "NAME": BASE_DIR / "db.sqlite3",
+    }
+}
+
+TASKS = {
+    "default": {
+        "BACKEND": "django_tasks_db.DatabaseBackend",
+        "QUEUES": ["default", "emails", "reports"],
+    },
+}
+
+EMAIL_BACKEND = "django.core.mail.backends.console.EmailBackend"
+DEFAULT_FROM_EMAIL = "noreply@example.com"
+
+USE_TZ = True
+TIME_ZONE = "UTC"
+DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"

django-tasks/config/settings.py

@@ -0,0 +1,5 @@
+from django.urls import include, path
+
+urlpatterns = [
+    path("", include("myapp.urls")),
+]

django-tasks/config/urls.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+"""Django's command-line utility for administrative tasks."""
+import os
+import sys
+
+
+def main():
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
+    try:
+        from django.core.management import execute_from_command_line
+    except ImportError as exc:
+        raise ImportError(
+            "Couldn't import Django. Are you sure it's installed and "
+            "available on your PYTHONPATH environment variable? Did you "
+            "forget to activate a virtual environment?"
+        ) from exc
+    execute_from_command_line(sys.argv)
+
+
+if __name__ == "__main__":
+    main()

django-tasks/manage.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class MyappConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "myapp"

django-tasks/myapp/__init__.py

@@ -0,0 +1,30 @@
+# Generated by Django 6.0.3 on 2026-05-03 11:08
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Order',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('item', models.CharField(default='widget', max_length=100)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='User',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('email', models.EmailField(max_length=254)),
+                ('name', models.CharField(blank=True, max_length=100)),
+            ],
+        ),
+    ]

django-tasks/myapp/apps.py

@@ -0,0 +1,17 @@
+from django.db import models
+
+
+class User(models.Model):
+    email = models.EmailField()
+    name = models.CharField(max_length=100, blank=True)
+
+    def __str__(self):
+        return self.email
+
+
+class Order(models.Model):
+    item = models.CharField(max_length=100, default="widget")
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    def __str__(self):
+        return f"Order {self.pk} ({self.item})"