Django Tasks: Exploring the Built-in Tasks Framework

Start Using Django Tasks

Before pulling in any third-party package, you can try the Tasks API end-to-end with what ships in the framework itself. Django 6.0 includes two task backends out of the box: ImmediateBackend and DummyBackend. Both exist for development and testing. ImmediateBackend runs each task in the calling thread the moment you call .enqueue(), so it lets you define tasks and confirm they work without spinning up a worker.

Set up a virtual environment and install Django before scaffolding the project:

$ python-mvenvvenv
$ sourcevenv/bin/activate
(venv) $ python-mpipinstall"django>=6.0,<7.0"

The first command creates a fresh virtual environment in a venv/ folder, and the second activates it. From this point on, every shell command in the tutorial assumes you’re inside the active venv, which is why each prompt is prefixed with (venv) $.

If you already have a Django version 6.0 project handy, you can extend it here. Otherwise, follow the Django setup guide to scaffold one with config/ and myapp/ directories. Then add the following to your settings module:

# ...

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

# ...

That single setting tells Django to dispatch every enqueued task through ImmediateBackend, which runs tasks inline in the current thread.

You’ll work with a small project throughout this tutorial. After django-admin startproject config . and python -m manage startapp myapp, you’ll have a standard layout: a config/ settings module that holds settings.py and urls.py, plus a myapp/ app for your code.

Beyond that layout, you’ll add a few files by hand. Inside myapp/, you’ll create tasks.py for the task definitions and urls.py to route the demo views you’ll build, and at the project root you’ll add a requirements.txt to pin dependencies. Make sure "myapp" appears in INSTALLED_APPS in config/settings.py so that Django picks up the tasks you define inside it.

startapp doesn’t create a tasks module for you, so create a new myapp/tasks.py file. By convention, the framework looks for a tasks module in each installed app:

fromdjango.tasksimport task

@task
defsay_hello(name):
    return f"Hello, {name}!"

The @task decorator wraps say_hello() in a Task object, which exposes .enqueue() and related methods for sending the function to a worker.

Now open the Django shell, which gives you a Python REPL with your project’s settings preloaded:

(venv) $ python-mmanageshell

Inside the shell, enqueue the function:

>>> frommyapp.tasksimport say_hello
>>> result = say_hello.enqueue("Real Python")
Task id=D32SVJEMf0fbRXS0yLp8vbfP9HSqs9zd path=myapp.tasks.say_hello
⮑ state=RUNNING
Task id=D32SVJEMf0fbRXS0yLp8vbfP9HSqs9zd path=myapp.tasks.say_hello
⮑ state=SUCCESSFUL
>>> result.status
TaskResultStatus.SUCCESSFUL
>>> result.return_value
'Hello, Real Python!'

Because ImmediateBackend runs the task inline in the same process as the shell, the framework’s task lifecycle log messages appear right inside your REPL session, and .status is already SUCCESSFUL by the time .enqueue() returns. That’s useful for sanity checks, but it isn’t how you want background work to behave in production. To run tasks in a real worker process, swap in a database-backed backend.

The community package django-tasks-db provides one. Install the package with python -m pip:

(venv) $ python-mpipinstalldjango-tasks-db

That command pulls django-tasks-db from PyPI into your active virtual environment, along with the django-tasks backport it depends on.

Now that you have both dependencies installed, capture them in a requirements.txt file at the top of your project so the setup is reproducible:

django>=6.0,<7.0
django-tasks-db==0.12.0

With your dependencies pinned, add django_tasks_db to INSTALLED_APPS and update your TASKS setting:

# ...

INSTALLED_APPS = [
    # ...
    "django_tasks_db",
]

# ...

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

# ...

Adding django_tasks_db to INSTALLED_APPS registers the migrations for the table that stores enqueued work. The TASKS entry tells Django to route the default backend through the database-backed worker. The three named queues will let you separate fast and slow jobs later.

With the backend wired up, run migrations to create the table:

(venv) $ python-mmanagemigrate

Running migrate creates the DBTaskResult table that django-tasks-db uses to persist enqueued and finished tasks.

Open two terminals. In the first, start the worker:

(venv) $ python-mmanagedb_worker
Starting worker worker_id=7mt39d9zuzUrgIb1laqQi2JRVuLx9Hqh queues=default

The worker now polls the database for ready tasks. In the second terminal, drop into the shell and enqueue your task:

>>> frommyapp.tasksimport say_hello
>>> result = say_hello.enqueue("Real Python")
>>> result.status
TaskResultStatus.READY
>>> result.refresh()
>>> result.status
TaskResultStatus.SUCCESSFUL
>>> result.return_value
'Hello, Real Python!'

This time, .status starts as READY because the task is sitting in the database waiting for the worker to pick it up. Within a second, the worker logs that the task ran:

Task id=95bc9615-7f61-4660-a433-f5a0e8120b01 path=myapp.tasks.say_hello
⮑ state=RUNNING
Task id=95bc9615-7f61-4660-a433-f5a0e8120b01 path=myapp.tasks.say_hello
⮑ state=SUCCESSFUL

A call to .refresh() on the result pulls the latest state from the database. With the worker handling execution, your task code is exactly the same, but it runs out of the way of any view that enqueues it.

Defer Work Outside the Request With @task

The @task decorator is the entry point for the whole framework. Apply it to any module-level function whose work you want to push out of the request cycle. The decorator wraps the function in a Task object that exposes .enqueue() for synchronous code. The official django.tasks documentation covers every attribute and method on the resulting object if you want the full reference later.

Now consider a more realistic example. A user signs up, and you want to send a welcome email without making them wait for the SMTP handshake. Add the new task alongside say_hello() in the same file. The highlighted lines below show what’s new compared to the previous version:

fromdjango.core.mailimport send_mail
fromdjango.tasksimport task

frommyapp.modelsimport User

@task
defsay_hello(name):
    return f"Hello, {name}!"

@task
defsend_welcome_email(user_id):
    user = User.objects.get(pk=user_id)
    send_mail(
        subject="Welcome!",
        message="Thanks for joining.",
        from_email=None,
        recipient_list=[user.email],
    )

Notice that the task takes a user_id rather than a user instance. Both task arguments and return values get serialized to JSON when the task is enqueued. Model instances, datetime values, and tuples used as dict keys won’t survive the round trip. Pass identifiers and look up the related objects inside the task body.

Because send_welcome_email() sends real email, point EMAIL_BACKEND at the console backend while you develop. That prints each message to your terminal instead of opening a connection to an SMTP server:

# ...

EMAIL_BACKEND = "django.core.mail.backends.console.EmailBackend"

# ...

You can call this task from a view the way you’d call any function, except you replace the direct call with .enqueue():

fromdjango.httpimport JsonResponse

frommyapp.tasksimport send_welcome_email

defregister(request):
    user = create_user(request)
    result = send_welcome_email.enqueue(user.id)
    return JsonResponse({"task_id": str(result.id)})

The view returns immediately with the task’s ID, while the worker handles the email in the background. The user sees a fast response, and your application server isn’t held up by an unreliable SMTP server.

The timeline below makes that trade-off concrete. It runs the same job both ways—inline in the view versus enqueued with @task—so you can watch where the user’s response lands in each case:

To verify the flow end-to-end, you need to route the /register/ URL to the view. startapp doesn’t create a URL conf for the app, so create a new myapp/urls.py file by hand:

fromdjango.urlsimport path

frommyappimport views

urlpatterns = [
    path("register/", views.register, name="register"),
]

Then include the app’s URL conf from the project’s root URL conf so Django can find it:

fromdjango.urlsimport include, path

urlpatterns = [
    path("", include("myapp.urls")),
]

With the URL wired up, you now need three concurrent processes to exercise the flow. In one terminal, start the worker:

(venv) $ python-mmanagedb_worker

In a second terminal, start the Django development server:

(venv) $ python-mmanagerunserver

In a third terminal, hit the endpoint with curl:

(venv) $ curl"http://localhost:8000/register/?email=test@example.com"
{"task_id": "d52e37ca-2b10-4a66-9ae3-eca99a3a54bd"}

The dev server returns the JSON response immediately. Meanwhile, in the worker terminal, you’ll see the task move through RUNNING and then SUCCESSFUL:

Task id=d52e37ca-2b10-4a66-9ae3-eca99a3a54bd path=myapp.tasks.send_welcome_email
⮑ state=RUNNING
Task id=d52e37ca-2b10-4a66-9ae3-eca99a3a54bd path=myapp.tasks.send_welcome_email
⮑ state=SUCCESSFUL

With the console backend, the welcome email prints between those two log lines in the same worker terminal. The task ID in the response lets a follow-up request check on the work after it finishes. Wiring up that request is the next thing you’ll do.

Check a Task’s Result After It Runs

Returning a task ID from a view is only useful if you can check on the task later. The framework gives you two ways to retrieve a result by ID. You can call .get_result() on the task object, or you can ask the default backend directly. Add a second view in the same myapp/views.py file, where the highlighted lines are new on top of the register() view from the previous section:

fromdjango.httpimport JsonResponse
fromdjango.tasksimport TaskResultStatus, default_task_backend

frommyapp.tasksimport send_welcome_email

defregister(request):
    ...

deftask_status(request, task_id):
    result = default_task_backend.get_result(task_id)
    result.refresh()
    if result.status == TaskResultStatus.SUCCESSFUL:
        return JsonResponse({"status": "done", "value": result.return_value})
    if result.is_finished:
        return JsonResponse({"status": "failed"})
    return JsonResponse({"status": "pending"})

A TaskResult object holds the task’s status as it was at the moment you fetched it. To pull the latest state from the database, call .refresh() in synchronous code or .arefresh() from an async view. The status moves through four values:

A finished task has either a SUCCESSFUL or FAILED status, so the .is_finished property is shorthand for when you don’t care which.

Don’t read .return_value on a task that hasn’t finished. The framework raises a ValueError if you try, which prevents you from acting on stale or absent data.

If a task fails, the exception class and traceback are preserved in the result for postmortem inspection. To see this in action, drop into the shell and enqueue send_welcome_email() with a user ID that doesn’t exist, then refresh the result once the worker has had a chance to run it:

>>> frommyapp.tasksimport send_welcome_email
>>> result = send_welcome_email.enqueue(99999)
>>> result.refresh()
>>> result.status
TaskResultStatus.FAILED
>>> result.errors
[TaskError(exception_class_path='myapp.models.User.DoesNotExist',
⮑ traceback='...')]
>>> result.errors[0].exception_class_path
'myapp.models.User.DoesNotExist'

The repr of the TaskError truncates the traceback to keep the output readable. To see the full stack, print the .traceback attribute:

>>> print(result.errors[0].traceback)
Traceback (most recent call last):
...
myapp.models.User.DoesNotExist: User matching query does not exist.

Together, the exception class path and the full traceback string are enough to log a meaningful error or show a friendly message to the user.

Separate Fast and Slow Work With Named Queues

By default, every task lands in a queue called default. This works for a small app, but the moment you mix latency-sensitive jobs like welcome emails with long-running ones like monthly report PDFs, you’ll want separate lanes.

The @task decorator accepts a queue_name and a priority for this purpose. Update the decorator on send_welcome_email() to route it to a dedicated emails queue, then add a new generate_monthly_report() task on a reports queue:

fromdjango.core.mailimport send_mail
fromdjango.tasksimport task

frommyapp.modelsimport User

@task
defsay_hello(name):
    return f"Hello, {name}!"

@task(queue_name="emails", priority=2)
defsend_welcome_email(user_id):
    ...

@task(queue_name="reports", priority=0)
defgenerate_monthly_report(month):
    return f"report for {month} ready"

Priority ranges from -100 to 100. Higher priorities run sooner within a queue, which is helpful when you want to bump time-sensitive jobs ahead of the rest of the queue.

Run a dedicated worker for each queue so the email worker doesn’t get stuck behind a report job that takes half an hour. Because each db_worker invocation runs in the foreground until you stop it, open one terminal per queue. In the first terminal, start the emails worker:

(venv) $ python-mmanagedb_worker--queue-nameemails

In a second terminal, start the reports worker:

(venv) $ python-mmanagedb_worker--queue-namereports

A dedicated worker per queue keeps time-sensitive tasks isolated from heavy background work, so a slow monthly report can’t delay a welcome email.

Beware of Limitations and Gotchas

The new framework lowers the bar for adding background work to your project, but a few sharp edges are worth knowing about before you ship:

• The built-in backends aren’t suitable for production: ImmediateBackend runs each task in the calling thread, so there’s no concurrency, no retry, and no way to inspect a task after the fact. DummyBackend goes further and never runs the task at all, leaving every result frozen at READY. Both are meant only for local development. For production, use a backend such as django-tasks-db or another adapter from the Django ecosystem page.
• django-tasks-db is community-maintained: It’s a separate package from the original django-tasks backport. Before adopting it, check the project’s GitHub activity, release cadence, and issue backlog. Production readiness is a moving target for any third-party dependency.
• Task arguments must be JSON-serializable: Pass user IDs instead of user instances, ISO strings instead of datetime objects, and lists instead of tuples used as dict keys. Serialization happens at enqueue time, so a hidden non-JSON value won’t surface until you try to enqueue.
• Enqueueing inside transaction.atomic() can occur before the commit: If a view creates a row and enqueues a task that needs that row, then a worker might pick up the task before the transaction commits and fail with a missing-record error. Wrap the enqueue call with transaction.on_commit to push it past the commit. See the snippet below for the recommended pattern.
• There’s no built-in scheduling: The framework runs tasks as soon as a worker is available to pick them up. If you need cron-style or recurring jobs, then add a scheduler such as django-crontask for periodic tasks, or use management commands triggered by system cron.

Here’s the transaction.on_commit pattern in action. Add a checkout() view to the same myapp/views.py file, with the highlighted lines showing what’s new:

fromfunctoolsimport partial

fromdjango.dbimport transaction
fromdjango.httpimport JsonResponse
fromdjango.tasksimport TaskResultStatus, default_task_backend
frommyapp.modelsimport Order
frommyapp.tasksimport process_order, send_welcome_email

defregister(request):
    ...

deftask_status(request, task_id):
    ...

defcheckout(request):
    with transaction.atomic():
        order = Order.objects.create()
        transaction.on_commit(partial(process_order.enqueue, order.id))
    return JsonResponse({"order_id": order.id})

Here, process_order() is another @task-decorated function defined just like the tasks earlier in this tutorial, and Order is an ordinary model. Wrapping .enqueue() in partial lets transaction.on_commit invoke it once the surrounding atomic block commits, which prevents the worker from acting on a row the database hasn’t saved yet.

Conclusion

Django Tasks gives you a standardized API for background work. You define a task with the @task decorator, call .enqueue() to send it to a queue, and check on its progress with a TaskResult.

The whole API is backend-agnostic, so the same task code works against the development backends or a production-grade adapter such as django-tasks-db. As the package ecosystem matures, you’ll be able to swap backends with a single setting change instead of rewriting your task layer.

In this tutorial, you’ve learned how to:

• Configure the built-in Tasks framework with both a development and a production backend
• Define and enqueue background tasks using the @task decorator
• Track task results across requests with .get_result(), .refresh(), and the status lifecycle
• Route work to dedicated queues using queue_name and priority
• Avoid the most common production pitfalls, including transaction races and JSON-only arguments

For a structured route through Django itself, Real Python’s Django for Web Development learning path has tutorials and video courses that take you from a first project to a deployed app. Otherwise, try out the new Tasks API in your next project and let the framework handle the queueing for you.

Frequently Asked Questions

Now that you’ve gained some experience with the new Tasks framework, you can use the questions and answers below to check your understanding and recap what you’ve learned.