Add django 6.1 support

django_q/cluster.py

@@ -31,7 +31,6 @@ from django_q.conf import (
     psutil,
     setproctitle,
 )
-from django_q.enums import TimerStatus
 from django_q.humanhash import humanize
 from django_q.monitor import monitor
 from django_q.pusher import pusher
@@ -235,11 +234,7 @@ class Sentinel:
 
     def spawn_worker(self):
         self.spawn_process(
-            worker,
-            self.task_queue,
-            self.result_queue,
-            Value("f", TimerStatus.IDLE),
-            self.timeout,
+            worker, self.task_queue, self.result_queue, Value("f", -1), self.timeout
         )
 
     def spawn_monitor(self) -> Process:
@@ -276,7 +271,7 @@ class Sentinel:
 
             self.pool.remove(process)
             self.spawn_worker()
-            if process.timer.value == TimerStatus.TIMEOUT:
+            if process.timer.value == 0:
                 # only need to terminate on timeout, otherwise we risk destabilizing
                 # the queues
                 task_name = ""
@@ -301,7 +296,7 @@ class Sentinel:
                         "name": process.name
                     }
                 logger.critical(msg)
-            elif int(process.timer.value) == TimerStatus.RECYCLED:
+            elif int(process.timer.value) == -2:
                 logger.info(_("recycled worker %(name)s") % {"name": process.name})
             else:
                 logger.critical(
@@ -353,11 +348,11 @@ class Sentinel:
             for p in self.pool:
                 with p.timer.get_lock():
                     # Are you alive?
-                    if not p.is_alive() or p.timer.value == TimerStatus.TIMEOUT:
+                    if not p.is_alive() or p.timer.value == 0:
                         self.reincarnate(p)
                         continue
                     # Decrement timer if work is being done
-                    if p.timer.value > TimerStatus.TIMEOUT:
+                    if p.timer.value > 0:
                         p.timer.value -= cycle
             # Check Monitor
             if not self.monitor.is_alive():

django_q/conf.py

@@ -7,7 +7,6 @@ from multiprocessing import cpu_count
 from warnings import warn
 
 from django.conf import settings
-from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext_lazy as _
 
 from django_q.queues import Queue
@@ -212,12 +211,7 @@ class Conf:
 
     # Use the secret key for package signing
     # Django itself should raise an error if it's not configured
-    # but suppress the exception early to allow other parts of the app to still run.
-    # For example any "python manage.py ..." command still runs even if SECRET_KEY is not set.
-    try:
-        SECRET_KEY = settings.SECRET_KEY
-    except ImproperlyConfigured:
-        SECRET_KEY = None
+    SECRET_KEY = settings.SECRET_KEY
 
     # The redis stats key
     Q_STAT = f"django_q:{PREFIX}:cluster"

django_q/enums.py

@@ -1,14 +0,0 @@
-from enum import IntEnum
-
-
-class TimerStatus(IntEnum):
-    """
-    Sentinel values for the countdown timer a worker shares with the sentinel.
-
-    Any positive value is the number of seconds left before the sentinel
-    considers the task timed out and reincarnates the worker.
-    """
-
-    TIMEOUT = 0  # task timed out, the worker has to be terminated
-    IDLE = -1  # waiting for work, or working without a timeout
-    RECYCLED = -2  # recycle limit reached, the worker stopped on its own

django_q/management/commands/qcluster.py

@@ -1,6 +1,5 @@
 import os
 
-from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.utils.translation import gettext as _
 
@@ -29,8 +28,6 @@ class Command(BaseCommand):
         )
 
     def handle(self, *args, **options):
-        # Ensure that the cluster starts only if the SECRET_KEY is set, as it is required for signing.
-        settings.SECRET_KEY
         # Set alternative cluster_name before creating the cluster (cluster_name is broker's queue_name, too)
         cluster_name = options.get("cluster_name")
         if cluster_name:

django_q/tests/test_cached.py

@@ -6,7 +6,6 @@ import pytest
 
 from django_q.brokers import get_broker
 from django_q.conf import Conf
-from django_q.enums import TimerStatus
 from django_q.monitor import monitor
 from django_q.pusher import pusher
 from django_q.queues import Queue
@@ -68,7 +67,7 @@ def test_cached(broker):
     assert task_queue.qsize() == task_count
     task_queue.put("STOP")
     result_queue = Queue()
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     assert result_queue.qsize() == task_count
     result_queue.put("STOP")
     monitor(result_queue)

django_q/tests/test_cluster.py

@@ -15,7 +15,6 @@ from django.utils import timezone
 from django_q.brokers import Broker, get_broker
 from django_q.cluster import Cluster, Sentinel
 from django_q.conf import Conf
-from django_q.enums import TimerStatus
 from django_q.humanhash import DEFAULT_WORDLIST, uuid
 from django_q.models import Success, Task
 from django_q.monitor import monitor, save_task
@@ -246,7 +245,7 @@ def test_cluster(broker):
     assert queue_size(broker=broker) == 0
     # Test work
     task_queue.put("STOP")
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     assert task_queue.qsize() == 0
     assert result_queue.qsize() == 1
     # Test monitor
@@ -273,7 +272,7 @@ def test_results(broker):
     pusher(task_queue, stop_event, broker=broker)
     task_queue.put("STOP")
     result_queue = Queue()
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     result_queue.put("STOP")
     monitor(result_queue)
 
@@ -373,7 +372,7 @@ def test_enqueue(broker, admin_user):
     assert fetch_group("test_j", count=2, wait=10) is None
     # let a worker handle them
     result_queue = Queue()
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     assert result_queue.qsize() == task_count
     result_queue.put("STOP")
     # store the results
@@ -542,7 +541,7 @@ def test_recycle(broker, monkeypatch, django_assert_num_queries):
     pusher(task_queue, stop_event, broker=broker)
     pusher(task_queue, stop_event, broker=broker)
     # worker should exit on recycle
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     # check if the work has been done
     assert result_queue.qsize() == 2
     # save_limit test
@@ -574,7 +573,7 @@ def test_save_limit_per_func(broker, monkeypatch):
     threading.Timer(3, stop_event.set).start()
     for i in range(3):
         pusher(task_queue, stop_event, broker=broker)
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker)
     assert start_event.is_set()
     assert s.status() == Conf.STOPPED
@@ -619,7 +618,7 @@ def test_max_rss(broker, monkeypatch):
     # push the task
     pusher(task_queue, stop_event, broker=broker)
     # worker should exit on recycle
-    worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("f", -1))
     # check if the work has been done
     assert result_queue.qsize() == 1
     # save_limit test
@@ -655,7 +654,7 @@ def test_bad_secret(broker, monkeypatch):
     worker(
         task_queue,
         result_queue,
-        Value("f", TimerStatus.IDLE),
+        Value("f", -1),
     )
     assert result_queue.qsize() == 0
     broker.delete_queue()
@@ -839,7 +838,7 @@ class TestSignals:
         event.set()
         pusher(task_queue, event, broker=broker)
         task_queue.put("STOP")
-        worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+        worker(task_queue, result_queue, Value("f", -1))
         result_queue.put("STOP")
         monitor(result_queue, broker)
         broker.delete_queue()
@@ -868,7 +867,7 @@ class TestSignals:
         event.set()
         pusher(task_queue, event, broker=broker)
         task_queue.put("STOP")
-        worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+        worker(task_queue, result_queue, Value("f", -1))
         result_queue.put("STOP")
         monitor(result_queue, broker)
         broker.delete_queue()
@@ -897,7 +896,7 @@ class TestSignals:
         event.set()
         pusher(task_queue, event, broker=broker)
         task_queue.put("STOP")
-        worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
+        worker(task_queue, result_queue, Value("f", -1))
         result_queue.put("STOP")
         monitor(result_queue, broker)
         broker.delete_queue()

django_q/tests/test_scheduler.py

@@ -13,7 +13,6 @@ from django.utils.timezone import is_naive
 
 from django_q.brokers import Broker, get_broker
 from django_q.conf import Conf
-from django_q.enums import TimerStatus
 from django_q.monitor import monitor
 from django_q.pusher import pusher
 from django_q.queues import Queue
@@ -223,7 +222,7 @@ def test_scheduler(broker, monkeypatch):
     task_queue.put("STOP")
     # let a worker handle them
     result_queue = Queue()
-    worker(task_queue, result_queue, Value("b", TimerStatus.IDLE))
+    worker(task_queue, result_queue, Value("b", -1))
     assert result_queue.qsize() == 1
     result_queue.put("STOP")
     # store the results

django_q/worker.py

@@ -17,7 +17,6 @@ except core.exceptions.AppRegistryNotReady:
     django.setup()
 
 from django_q.conf import Conf, error_reporter, logger, resource, setproctitle
-from django_q.enums import TimerStatus
 from django_q.exceptions import TimeoutException
 from django_q.signals import post_execute_in_worker, post_spawn, pre_execute
 from django_q.timeout import TimeoutHandler
@@ -55,11 +54,11 @@ def worker(
         setproctitle.setproctitle(f"qcluster {proc_name} idle")
     task_count = 0
     if timeout is None:
-        timeout = TimerStatus.IDLE
+        timeout = -1
     # Start reading the task queue
     for task in iter(task_queue.get, "STOP"):
         result = None
-        timer.value = TimerStatus.IDLE
+        timer.value = -1  # Idle
         task_count += 1
         f = task["func"]
 
@@ -93,7 +92,7 @@ def worker(
         pre_execute.send(sender="django_q", func=f, task=task)
         # execute the payload
         timer.value = timer_value  # Busy
-        if timer.value != TimerStatus.IDLE:
+        if timer.value != -1:
             timer.value += 3  # Add buffer so that guard doesn't kill the process on timeout before it gets processed
 
         timeout_error = False
@@ -123,15 +122,15 @@ def worker(
             result_queue.put(task)
             if timeout_error:
                 # force destroy process due to timeout
-                timer.value = TimerStatus.TIMEOUT
+                timer.value = 0
                 break
 
-            timer.value = TimerStatus.IDLE
+            timer.value = -1  # Idle
             if setproctitle:
                 setproctitle.setproctitle(f"qcluster {proc_name} idle")
             # Recycle
             if task_count == Conf.RECYCLE or rss_check():
-                timer.value = TimerStatus.RECYCLED
+                timer.value = -2  # Recycled
                 break
     logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name})