Add the exclude_future_jobs arg to the queue_depth command

The queue_depth command can call get_queue_depths with the option to exclude jobs in the future

Author: George9Waller

django_dbq/management/commands/queue_depth.py

@@ -8,10 +8,13 @@ class Command(BaseCommand):
 
     def add_arguments(self, parser):
         parser.add_argument("queue_name", nargs="*", default=["default"], type=str)
+        parser.add_argument("--exclude_future_jobs", default=False, type=bool)
 
     def handle(self, *args, **options):
         queue_names = options["queue_name"]
-        queue_depths = Job.get_queue_depths()
+        queue_depths = Job.get_queue_depths(
+            exclude_future_jobs=options["exclude_future_jobs"]
+        )
 
         queue_depths_string = " ".join(
             [

django_dbq/tests.py

@@ -111,15 +111,19 @@ def test_get_queue_depths_exclude_future_jobs(self):
         self.assertDictEqual(queue_depths, {"default": 1, "testworker": 1})
 
 
+@freezegun.freeze_time("2025-01-01T12:00:00Z")
 @override_settings(JOBS={"testjob": {"tasks": ["a"]}})
 class QueueDepthTestCase(TestCase):
     def test_queue_depth(self):
-
         Job.objects.create(name="testjob", state=Job.STATES.FAILED)
         Job.objects.create(name="testjob", state=Job.STATES.NEW)
         Job.objects.create(name="testjob", state=Job.STATES.FAILED)
         Job.objects.create(name="testjob", state=Job.STATES.COMPLETE)
-        Job.objects.create(name="testjob", state=Job.STATES.READY)
+        Job.objects.create(
+            name="testjob",
+            state=Job.STATES.READY,
+            run_after=timezone.make_aware(datetime(2025, 1, 1, 13, 0, 0)),
+        )
         Job.objects.create(
             name="testjob", queue_name="testqueue", state=Job.STATES.READY
         )
@@ -132,6 +136,28 @@ def test_queue_depth(self):
         output = stdout.getvalue()
         self.assertEqual(output.strip(), "event=queue_depths default=2")
 
+    def test_queue_depth_exclude_future_jobs(self):
+        Job.objects.create(name="testjob", state=Job.STATES.FAILED)
+        Job.objects.create(name="testjob", state=Job.STATES.NEW)
+        Job.objects.create(name="testjob", state=Job.STATES.FAILED)
+        Job.objects.create(name="testjob", state=Job.STATES.COMPLETE)
+        Job.objects.create(
+            name="testjob",
+            state=Job.STATES.READY,
+            run_after=timezone.make_aware(datetime(2025, 1, 1, 13, 0, 0)),
+        )
+        Job.objects.create(
+            name="testjob", queue_name="testqueue", state=Job.STATES.READY
+        )
+        Job.objects.create(
+            name="testjob", queue_name="testqueue", state=Job.STATES.READY
+        )
+
+        stdout = StringIO()
+        call_command("queue_depth", exclude_future_jobs=True, stdout=stdout)
+        output = stdout.getvalue()
+        self.assertEqual(output.strip(), "event=queue_depths default=1")
+
     def test_queue_depth_multiple_queues(self):
 
         Job.objects.create(name="testjob", state=Job.STATES.FAILED)