Experiment with tasks panel

Author: Chiemezuo

debug_toolbar/panels/tasks.py

@@ -33,7 +33,55 @@ def generate_stats(self, request, response):
         self.record_stats(stats)
 
     def enable_instrumentation(self):
-        pass
+        """Hook into task system to collect queued tasks"""
+        try:
+            import django
+
+            if django.VERSION < (6, 0):
+                return
+            from django.tasks import Task
+
+            print("[TasksPanel] instrumentation enabled:", hasattr(Task, "enqueue"))
+
+            # Store original enqueue method
+            if hasattr(Task, "enqueue"):
+                self._original_enqueue = Task.enqueue
+
+                def wrapped_enqueue(task, *args, **kwargs):
+                    result = self._original_enqueue(task, *args, **kwargs).return_value
+                    self._record_task(task, args, kwargs, result)
+                    return result
+
+                Task.enqueue = wrapped_enqueue
+        except (ImportError, AttributeError):
+            pass
+
+    def _record_task(self, task, args, kwargs, result):
+        """Record a task that was queued"""
+        task_info = {
+            "name": getattr(task, "__name__", str(task)),
+            "args": repr(args) if args else "",
+            "kwargs": repr(kwargs) if kwargs else "",
+        }
+        self.queued_tasks.append(task_info)
 
     def disable_instrumentation(self):
-        pass
+        """Restore original methods"""
+        try:
+            from django.tasks import Task
+
+            if hasattr(self, "_original_enqueue"):
+                Task.enqueue = self._original_enqueue
+        except (ImportError, AttributeError):
+            pass
+
+    def _check_tasks_available(self):
+        """Check if Django tasks system is available"""
+        try:
+            import django
+
+            if django.VERSION < (6, 0):
+                return False
+            return True
+        except (ImportError, AttributeError):
+            return False

example/async_/tasks.py

@@ -0,0 +1,19 @@
+try:
+    from django.tasks import task
+except ImportError:
+    # Define a fallback decorator
+    def task(func=None, **kwargs):
+        def decorator(f):
+            return f
+
+        return decorator if func is None else decorator(func)
+
+
+@task
+def send_welcome_message(message):
+    return f"Sent message: {message}"
+
+
+@task
+def generate_report(report_id):
+    return f"Report {report_id} generated"

example/views.py

@@ -1,5 +1,6 @@
 import asyncio
 
+import django
 from asgiref.sync import sync_to_async
 from django.contrib.auth.models import User
 from django.http import JsonResponse
@@ -20,6 +21,13 @@ def jinja2_view(request):
 
 
 async def async_home(request):
+    if django.VERSION >= (6, 0):
+        from .async_.tasks import generate_report, send_welcome_message
+
+        # Queue some tasks
+        send_welcome_message.enqueue(message="hi there")
+        generate_report.enqueue(report_id=456)
+
     return await sync_to_async(render)(request, "index.html")