Add exclude_future_jobs arg to get_queue_depths

New arg exclude_future_jobs filters the READY and NEW jobs to only be ones without a run_after or the run_after is now/in the past

Author: George9Waller

django_dbq/models.py

@@ -8,7 +8,7 @@
     get_failure_hook_name,
     get_creation_hook_name,
 )
-from django.db.models import JSONField, UUIDField, Count, TextChoices
+from django.db.models import JSONField, UUIDField, Count, TextChoices, Q
 import datetime
 import logging
 import uuid
@@ -173,10 +173,17 @@ def run_creation_hook(self):
             creation_hook_function(self)
 
     @staticmethod
-    def get_queue_depths():
+    def get_queue_depths(*, exclude_future_jobs=False):
+        jobs_waiting_in_queue = Job.objects.filter(
+            state__in=(Job.STATES.READY, Job.STATES.NEW)
+        )
+        if exclude_future_jobs:
+            jobs_waiting_in_queue = jobs_waiting_in_queue.filter(
+                Q(run_after__isnull=True) | Q(run_after__lte=timezone.now())
+            )
+
         annotation_dicts = (
-            Job.objects.filter(state__in=(Job.STATES.READY, Job.STATES.NEW))
-            .values("queue_name")
+            jobs_waiting_in_queue.values("queue_name")
             .order_by("queue_name")
             .annotate(Count("queue_name"))
         )

django_dbq/tests.py

@@ -71,12 +71,17 @@ def test_worker_with_queue_name(self):
         self.assertTrue("test_queue" in output)
 
 
+@freezegun.freeze_time("2025-01-01T12:00:00Z")
 @override_settings(JOBS={"testjob": {"tasks": ["a"]}})
 class JobModelMethodTestCase(TestCase):
     def test_get_queue_depths(self):
         Job.objects.create(name="testjob", queue_name="default")
         Job.objects.create(name="testjob", queue_name="testworker")
-        Job.objects.create(name="testjob", queue_name="testworker")
+        Job.objects.create(
+            name="testjob",
+            queue_name="testworker",
+            run_after=timezone.make_aware(datetime(2025, 1, 1, 13, 0, 0)),
+        )
         Job.objects.create(
             name="testjob", queue_name="testworker", state=Job.STATES.FAILED
         )
@@ -87,6 +92,24 @@ def test_get_queue_depths(self):
         queue_depths = Job.get_queue_depths()
         self.assertDictEqual(queue_depths, {"default": 1, "testworker": 2})
 
+    def test_get_queue_depths_exclude_future_jobs(self):
+        Job.objects.create(name="testjob", queue_name="default")
+        Job.objects.create(name="testjob", queue_name="testworker")
+        Job.objects.create(
+            name="testjob",
+            queue_name="testworker",
+            run_after=timezone.make_aware(datetime(2025, 1, 1, 13, 0, 0)),
+        )
+        Job.objects.create(
+            name="testjob", queue_name="testworker", state=Job.STATES.FAILED
+        )
+        Job.objects.create(
+            name="testjob", queue_name="testworker", state=Job.STATES.COMPLETE
+        )
+
+        queue_depths = Job.get_queue_depths(exclude_future_jobs=True)
+        self.assertDictEqual(queue_depths, {"default": 1, "testworker": 1})
+
 
 @override_settings(JOBS={"testjob": {"tasks": ["a"]}})
 class QueueDepthTestCase(TestCase):