Add Tasks panel showing tasks queued during the request

debug_toolbar/_compat.py

@@ -18,3 +18,20 @@ def login_not_required(view_func):
     # For Django < 6.0, there is no native CSP support, hence no CSP nonces.
     def get_nonce(request):
         return None
+
+
+# django.tasks was added in Django 6.0.
+django_has_tasks_support = django.VERSION >= (6, 0)
+
+if django_has_tasks_support:
+    from django.tasks import task
+else:
+    # For Django < 6.0, django.tasks doesn't exist. This pass-through
+    # decorator lets call sites decorate funtcions unconditionally
+    # tests that need real task behavior should use
+    # @skipUnless(django_has_tasks_support, ...) instead of relying on
+    # this decorator doing anything useful.
+    def task(func=None, **kwargs):
+        if func is None:
+            return lambda f: f
+        return func

debug_toolbar/panels/tasks.py

@@ -0,0 +1,81 @@
+from django import VERSION
+from django.utils.translation import gettext_lazy as _, ngettext
+
+from debug_toolbar.panels import Panel
+from debug_toolbar.utils import sanitize_and_sort_request_vars
+
+if VERSION >= (6, 0):
+    from django.tasks.signals import task_enqueued
+else:
+    task_enqueued = None
+
+
+class TasksPanel(Panel):
+    """
+    Panel that displays the tasks queued during the request.
+
+    This relies on Django's built-in tasks framework (``django.tasks``),
+    which was added in Django 6.0. On older versions of Django, the panel
+    explains that upgrading Django is required in order to see task
+    information.
+    """
+
+    template = "debug_toolbar/panels/tasks.html"
+
+    is_async = True
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.tasks: list = []  # populated with TaskResult instances in _record_task
+
+    nav_title = _("Tasks")
+
+    @property
+    def nav_subtitle(self):
+        if VERSION < (6, 0):
+            return _("Requires Django 6.0+")
+        tasks = self.get_stats()["tasks"]
+        return ngettext("%(count)d task", "%(count)d tasks", len(tasks)) % {
+            "count": len(tasks)
+        }
+
+    title = _("Tasks")
+
+    def _record_task(self, sender, task_result, **kwargs):
+        # Store the TaskResult as-is; it's only flattened into a
+        # JSON-serializable dict in generate_stats, right before the
+        # panel's stats get serialized (see debug_toolbar/store.py).
+        self.tasks.append(task_result)
+
+    def enable_instrumentation(self):
+        if task_enqueued is not None:
+            task_enqueued.connect(self._record_task)
+
+    def disable_instrumentation(self):
+        if task_enqueued is not None:
+            task_enqueued.disconnect(self._record_task)
+
+    def generate_stats(self, request, response):
+        tasks = []
+        for task_result in self.tasks:
+            task = task_result.task
+            tasks.append(
+                {
+                    "id": task_result.id,
+                    "module_path": task.module_path,
+                    "queue_name": task.queue_name,
+                    "priority": task.priority,
+                    "backend": task_result.backend,
+                    "run_after": task.run_after,
+                    "takes_context": task.takes_context,
+                    "args": sanitize_and_sort_request_vars(task_result.args),
+                    "kwargs": sanitize_and_sort_request_vars(task_result.kwargs),
+                    "status": task_result.status,
+                }
+            )
+        self.record_stats(
+            {
+                "tasks_available": VERSION >= (6, 0),
+                "tasks": tasks,
+            }
+        )

debug_toolbar/settings.py

@@ -84,6 +84,7 @@ def get_config():
     "debug_toolbar.panels.alerts.AlertsPanel",
     "debug_toolbar.panels.cache.CachePanel",
     "debug_toolbar.panels.signals.SignalsPanel",
+    "debug_toolbar.panels.tasks.TasksPanel",
     "debug_toolbar.panels.community.CommunityPanel",
     "debug_toolbar.panels.redirects.RedirectsPanel",
     "debug_toolbar.panels.profiling.ProfilingPanel",

debug_toolbar/templates/debug_toolbar/panels/tasks.html

@@ -0,0 +1,48 @@
+{% load i18n %}
+{% if tasks %}
+  <table>
+    <thead>
+      <tr>
+        <th>{% translate "python path"|title context "noun" %}</th>
+        <th>{% translate "queue"|title context "noun" %}</th>
+        <th>{% translate "backend"|title %}</th>
+        <th>{% translate "priority"|title %}</th>
+        <th>{% translate "run after"|title %}</th>
+        <th>{% translate "status"|title %}</th>
+        <th>{% translate "arguments"|title %}</th>
+        <th>{% translate "keyword arguments"|title %}</th>
+      </tr>
+    </thead>
+    <tbody>
+      {% for task in tasks %}
+        <tr>
+          <td>{{ task.module_path|escape }}</td>
+          <td>{{ task.queue_name|escape }}</td>
+          <td>{{ task.backend|escape }}</td>
+          <td>{{ task.priority }}</td>
+          <td>{{ task.run_after|default:"—" }}</td>
+          <td>{{ task.status }}</td>
+          <td><code>{{ task.args.raw|pprint }}</code></td>
+          <td>
+            <code>
+              {% if task.kwargs.list %}
+                {% for key, value in task.kwargs.list %}{{ key }}={{ value|pprint }}{% if not forloop.last %}, {% endif %}{% endfor %}
+              {% else %}
+                {{ task.kwargs.raw|pprint }}
+              {% endif %}
+            </code>
+          </td>
+        </tr>
+      {% endfor %}
+    </tbody>
+  </table>
+{% elif tasks_available %}
+  <p>{% translate "No tasks were queued during this request." %}</p>
+{% else %}
+  <p>
+    {% blocktranslate %}
+      The Tasks panel requires Django 6.0 or later. Upgrade Django to see
+      information about tasks queued during this request.
+    {% endblocktranslate %}
+  </p>
+{% endif %}

docs/changes.rst

@@ -2,8 +2,11 @@ Change log
 ==========
 
 Pending
--------
 
+
+* Added a Tasks panel that shows tasks queued during the request via
+  Django's built-in tasks framework (``django.tasks``, Django 6.0+). On
+  older versions of Django, the panel explains that upgrading is required.
 * Fixed the Django version check in the SQL panel test suite for Django's
   boolean parameter handling.
 

docs/panels.rst

@@ -108,6 +108,17 @@ Signals
 
 List of signals and receivers.
 
+Tasks
+~~~~~
+
+.. class:: debug_toolbar.panels.tasks.TasksPanel
+
+Shows the tasks queued during the request using Django's built-in tasks
+framework (``django.tasks``).
+
+.. note::
+   Requires Django 6.0+.
+
 Community
 ~~~~~~~~~
 

example/templates/index.html

@@ -16,6 +16,7 @@ <h1>Index of Tests</h1>
         <li><a href="{% url 'turbo' %}">Hotwire Turbo</a></li>
         <li><a href="{% url 'htmx' %}">htmx</a></li>
         <li><a href="{% url 'cache' %}">Cache</a></li>
+        <li><a href="{% url 'tasks' %}">Tasks</a></li>
         <li><a href="{% url 'bad_form' %}">Bad form</a></li>
       </ul>
       <p><a href="/admin/">Django Admin</a></p>

example/templates/tasks.html

@@ -0,0 +1,18 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <meta http-equiv="content-type" content="text/html; charset=utf-8">
+    <title>Tasks</title>
+  </head>
+  <body>
+    <h1>Tasks Test</h1>
+    {% if tasks_available %}
+      <p>Two tasks were just enqueued. Check the Tasks panel to see them.</p>
+    {% else %}
+      <p>
+        django.tasks requires Django 6.0+. Check the Tasks panel to see the
+        upgrade message.
+      </p>
+    {% endif %}
+  </body>
+</html>

example/urls.py

@@ -10,6 +10,7 @@
     cache_view,
     increment,
     jinja2_view,
+    tasks_view,
 )
 
 urlpatterns = [
@@ -49,6 +50,7 @@
         name="turbo2",
     ),
     path("cache/", cache_view, name="cache"),
+    path("tasks/", tasks_view, name="tasks"),
     path("admin/", admin.site.urls),
     path("ajax/increment", increment, name="ajax_increment"),
 ] + debug_toolbar_urls()

example/views.py

@@ -1,11 +1,28 @@
 import asyncio
 
+import django
 from asgiref.sync import sync_to_async
 from django.contrib.auth.models import User
 from django.core.cache import cache
 from django.http import JsonResponse
 from django.shortcuts import render
 
+TASKS_AVAILABLE = django.VERSION >= (6, 0)
+
+if TASKS_AVAILABLE:
+    from django.tasks import task
+
+    @task
+    def example_task(x=1, y=2):
+        return x + y
+
+
+def tasks_view(request):
+    if TASKS_AVAILABLE:
+        example_task.enqueue(1, y=2)
+        example_task.enqueue(3, y=4)
+    return render(request, "tasks.html", {"tasks_available": TASKS_AVAILABLE})
+
 
 def increment(request):
     try:

tests/panels/test_history.py

@@ -81,6 +81,7 @@ class HistoryViewsTestCase(IntegrationTestCase):
         "AlertsPanel",
         "CachePanel",
         "SignalsPanel",
+        "TasksPanel",
         "CommunityPanel",
         "ProfilingPanel",
     }

tests/panels/test_tasks.py

@@ -0,0 +1,89 @@
+import unittest
+
+from debug_toolbar._compat import django_has_tasks_support, task
+from debug_toolbar.panels.tasks import TasksPanel
+
+from ..base import BaseTestCase
+
+
+@task
+def sample_task(x, y=1):
+    return x + y
+
+
+class TasksPanelTestCase(BaseTestCase):
+    panel_id = TasksPanel.panel_id
+
+    @unittest.skipUnless(not django_has_tasks_support, "Requires Django < 6.0")
+    def test_nav_subtitle_without_tasks_support(self):
+        """
+        On Django < 6.0, django.tasks doesn't exist, so the panel should
+        explain that instead of showing task data.
+        """
+        self.panel.generate_stats(self.request, None)
+        self.assertEqual(str(self.panel.nav_subtitle), "Requires Django 6.0+")
+        stats = self.panel.get_stats()
+        self.assertEqual(stats, {"tasks_available": False, "tasks": []})
+
+    @unittest.skipUnless(django_has_tasks_support, "Requires Django 6.0+")
+    def test_no_tasks_queued(self):
+        self.panel.generate_stats(self.request, None)
+        stats = self.panel.get_stats()
+        self.assertEqual(stats, {"tasks_available": True, "tasks": []})
+        self.assertEqual(str(self.panel.nav_subtitle), "0 tasks")
+
+    @unittest.skipUnless(django_has_tasks_support, "Requires Django 6.0+")
+    def test_records_queued_task(self):
+        sample_task.enqueue(2, y=3)
+
+        self.panel.generate_stats(self.request, None)
+        stats = self.panel.get_stats()
+        self.assertEqual(len(stats["tasks"]), 1)
+
+        # id, priority, run_after, and status are runtime-dependent, so
+        # pull them from the actual result and assert the rest of the
+        # dict matches exactly around them.
+        recorded = stats["tasks"][0]
+        self.assertEqual(
+            stats,
+            {
+                "tasks_available": True,
+                "tasks": [
+                    {
+                        "id": recorded["id"],
+                        "module_path": f"{__name__}.sample_task",
+                        "queue_name": "default",
+                        "priority": recorded["priority"],
+                        "backend": "default",
+                        "run_after": recorded["run_after"],
+                        "takes_context": False,
+                        "args": {"raw": [2]},
+                        "kwargs": {"list": [("y", 3)]},
+                        "status": recorded["status"],
+                    }
+                ],
+            },
+        )
+
+    @unittest.skipUnless(django_has_tasks_support, "Requires Django 6.0+")
+    def test_nav_subtitle_counts_multiple_tasks(self):
+        sample_task.enqueue(1)
+        sample_task.enqueue(2)
+
+        self.panel.generate_stats(self.request, None)
+
+        self.assertEqual(str(self.panel.nav_subtitle), "2 tasks")
+
+    @unittest.skipUnless(django_has_tasks_support, "Requires Django 6.0+")
+    def test_disable_instrumentation_stops_recording(self):
+        self.panel.disable_instrumentation()
+        try:
+            sample_task.enqueue(1)
+        finally:
+            # Restore instrumentation so tearDown's disable call is a no-op
+            # rather than raising for double-disconnecting.
+            self.panel.enable_instrumentation()
+
+        self.panel.generate_stats(self.request, None)
+        stats = self.panel.get_stats()
+        self.assertEqual(stats["tasks"], [])