fix(admin): use redis.from_url() consistently for queue inspection and purge

- Replaced Kombu channel.client with redis.from_url() in both
  get_celery_queue_details() and purge_celery_queue_by_task_name()
  so both functions use the same connection mechanism
- Batched pipeline approach for per-task purge to avoid hitting
  Valkey's max query buffer limit on large queues

Author: valentijnscholten

dojo/utils.py

@@ -1233,41 +1233,40 @@ def get_celery_queue_details():
     """Per-task breakdown of the celery queue. O(N) — expensive for large queues."""
     tasks = {}
     try:
-        with Connection(settings.CELERY_BROKER_URL) as conn, conn.channel() as channel:
-            messages_raw = channel.client.lrange("celery", 0, -1)
-            for i, msg_raw in enumerate(messages_raw):
-                try:
-                    msg = json.loads(msg_raw)
-                    headers = msg.get("headers", {})
-                    task_name = headers.get("task", "unknown")
-                    eta = headers.get("eta")
-                    expires = headers.get("expires")
-                except Exception:
-                    task_name = "unknown"
-                    eta = None
-                    expires = None
-                if task_name not in tasks:
-                    tasks[task_name] = {
-                        "task_name": task_name,
-                        "count": 0,
-                        "oldest_position": i + 1,
-                        "newest_position": i + 1,
-                        "oldest_eta": eta,
-                        "newest_eta": eta,
-                        "earliest_expires": expires,
-                        "latest_expires": expires,
-                    }
-                else:
-                    tasks[task_name]["newest_position"] = i + 1
-                    tasks[task_name]["newest_eta"] = eta
-                    # ISO 8601 strings are lexicographically sortable
-                    if expires is not None:
-                        cur = tasks[task_name]
-                        if cur["earliest_expires"] is None or expires < cur["earliest_expires"]:
-                            cur["earliest_expires"] = expires
-                        if cur["latest_expires"] is None or expires > cur["latest_expires"]:
-                            cur["latest_expires"] = expires
-                tasks[task_name]["count"] += 1
+        client = redis_lib.from_url(settings.CELERY_BROKER_URL)
+        for i, msg_raw in enumerate(client.lrange("celery", 0, -1)):
+            try:
+                msg = json.loads(msg_raw)
+                headers = msg.get("headers", {})
+                task_name = headers.get("task", "unknown")
+                eta = headers.get("eta")
+                expires = headers.get("expires")
+            except Exception:
+                task_name = "unknown"
+                eta = None
+                expires = None
+            if task_name not in tasks:
+                tasks[task_name] = {
+                    "task_name": task_name,
+                    "count": 0,
+                    "oldest_position": i + 1,
+                    "newest_position": i + 1,
+                    "oldest_eta": eta,
+                    "newest_eta": eta,
+                    "earliest_expires": expires,
+                    "latest_expires": expires,
+                }
+            else:
+                tasks[task_name]["newest_position"] = i + 1
+                tasks[task_name]["newest_eta"] = eta
+                # ISO 8601 strings are lexicographically sortable
+                if expires is not None:
+                    cur = tasks[task_name]
+                    if cur["earliest_expires"] is None or expires < cur["earliest_expires"]:
+                        cur["earliest_expires"] = expires
+                    if cur["latest_expires"] is None or expires > cur["latest_expires"]:
+                        cur["latest_expires"] = expires
+            tasks[task_name]["count"] += 1
     except Exception:
         return None
     return sorted(tasks.values(), key=operator.itemgetter("oldest_position"))