Compare commits

...
Author SHA1 Message Date
Stan a8f3bfb832 Migrate timer status to enum instead of int 2026-08-13 16:53:42 +02:00
tugofpeasandTug of Peas 97ea3c89a5 Postpone SECRET_KEY evaluation to the qcluster command execution (#332)
* Suppress ImproperlyConfigured within Conf, allowing the exception to be raised specifically during the 'qcluster' command instead.

This ensures build scripts and workflows can freely execute other Django commands (e.g., 'python manage.py collectstatic') without failing due to a missing/empty secret key.

* Closes #279

Co-authored-by: Tug of Peas <19413623+tugofpeas@users.noreply.github.com>
2026-08-09 03:20:51 +02:00
Stan Triepels b2651ce86c Add django 6.1 support (#340) 2026-08-06 03:00:32 +02:00
11 changed files with 65 additions and 27 deletions
+6
View File
@@ -35,12 +35,18 @@ jobs:
django: django:
- "5.2" - "5.2"
- "6.0" - "6.0"
- "6.1"
exclude: exclude:
# django 6.0 does not support earlier than 3.12 # django 6.0 does not support earlier than 3.12
- python-version: "3.10" - python-version: "3.10"
django: "6.0" django: "6.0"
- python-version: "3.11" - python-version: "3.11"
django: "6.0" django: "6.0"
# django 6.1 does not support earlier than 3.12
- python-version: "3.10"
django: "6.1"
- python-version: "3.11"
django: "6.1"
services: services:
mongodb: mongodb:
+2 -2
View File
@@ -35,13 +35,13 @@ See the `changelog <https://github.com/GDay/django-q2/blob/master/CHANGELOG.md>`
Requirements Requirements
~~~~~~~~~~~~ ~~~~~~~~~~~~
- `Django <https://www.djangoproject.com>`__ 5.2 and 6.0 - `Django <https://www.djangoproject.com>`__ 5.2, 6.0 and 6.1
- `Django-picklefield <https://github.com/gintas/django-picklefield>`__ - `Django-picklefield <https://github.com/gintas/django-picklefield>`__
Tested with: Tested with:
* Python 3.10 to 3.14. * Python 3.10 to 3.14.
* Django 5.2 and 6.0. * Django 5.2, 6.0 and 6.1.
Brokers Brokers
~~~~~~~ ~~~~~~~
+10 -5
View File
@@ -31,6 +31,7 @@ from django_q.conf import (
psutil, psutil,
setproctitle, setproctitle,
) )
from django_q.enums import TimerStatus
from django_q.humanhash import humanize from django_q.humanhash import humanize
from django_q.monitor import monitor from django_q.monitor import monitor
from django_q.pusher import pusher from django_q.pusher import pusher
@@ -234,7 +235,11 @@ class Sentinel:
def spawn_worker(self): def spawn_worker(self):
self.spawn_process( self.spawn_process(
worker, self.task_queue, self.result_queue, Value("f", -1), self.timeout worker,
self.task_queue,
self.result_queue,
Value("f", TimerStatus.IDLE),
self.timeout,
) )
def spawn_monitor(self) -> Process: def spawn_monitor(self) -> Process:
@@ -271,7 +276,7 @@ class Sentinel:
self.pool.remove(process) self.pool.remove(process)
self.spawn_worker() self.spawn_worker()
if process.timer.value == 0: if process.timer.value == TimerStatus.TIMEOUT:
# only need to terminate on timeout, otherwise we risk destabilizing # only need to terminate on timeout, otherwise we risk destabilizing
# the queues # the queues
task_name = "" task_name = ""
@@ -296,7 +301,7 @@ class Sentinel:
"name": process.name "name": process.name
} }
logger.critical(msg) logger.critical(msg)
elif int(process.timer.value) == -2: elif int(process.timer.value) == TimerStatus.RECYCLED:
logger.info(_("recycled worker %(name)s") % {"name": process.name}) logger.info(_("recycled worker %(name)s") % {"name": process.name})
else: else:
logger.critical( logger.critical(
@@ -348,11 +353,11 @@ class Sentinel:
for p in self.pool: for p in self.pool:
with p.timer.get_lock(): with p.timer.get_lock():
# Are you alive? # Are you alive?
if not p.is_alive() or p.timer.value == 0: if not p.is_alive() or p.timer.value == TimerStatus.TIMEOUT:
self.reincarnate(p) self.reincarnate(p)
continue continue
# Decrement timer if work is being done # Decrement timer if work is being done
if p.timer.value > 0: if p.timer.value > TimerStatus.TIMEOUT:
p.timer.value -= cycle p.timer.value -= cycle
# Check Monitor # Check Monitor
if not self.monitor.is_alive(): if not self.monitor.is_alive():
+7 -1
View File
@@ -7,6 +7,7 @@ from multiprocessing import cpu_count
from warnings import warn from warnings import warn
from django.conf import settings from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from django_q.queues import Queue from django_q.queues import Queue
@@ -211,7 +212,12 @@ class Conf:
# Use the secret key for package signing # Use the secret key for package signing
# Django itself should raise an error if it's not configured # Django itself should raise an error if it's not configured
SECRET_KEY = settings.SECRET_KEY # 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
# The redis stats key # The redis stats key
Q_STAT = f"django_q:{PREFIX}:cluster" Q_STAT = f"django_q:{PREFIX}:cluster"
+14
View File
@@ -0,0 +1,14 @@
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
+3
View File
@@ -1,5 +1,6 @@
import os import os
from django.conf import settings
from django.core.management.base import BaseCommand from django.core.management.base import BaseCommand
from django.utils.translation import gettext as _ from django.utils.translation import gettext as _
@@ -28,6 +29,8 @@ class Command(BaseCommand):
) )
def handle(self, *args, **options): 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) # Set alternative cluster_name before creating the cluster (cluster_name is broker's queue_name, too)
cluster_name = options.get("cluster_name") cluster_name = options.get("cluster_name")
if cluster_name: if cluster_name:
+2 -1
View File
@@ -6,6 +6,7 @@ import pytest
from django_q.brokers import get_broker from django_q.brokers import get_broker
from django_q.conf import Conf from django_q.conf import Conf
from django_q.enums import TimerStatus
from django_q.monitor import monitor from django_q.monitor import monitor
from django_q.pusher import pusher from django_q.pusher import pusher
from django_q.queues import Queue from django_q.queues import Queue
@@ -67,7 +68,7 @@ def test_cached(broker):
assert task_queue.qsize() == task_count assert task_queue.qsize() == task_count
task_queue.put("STOP") task_queue.put("STOP")
result_queue = Queue() result_queue = Queue()
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
assert result_queue.qsize() == task_count assert result_queue.qsize() == task_count
result_queue.put("STOP") result_queue.put("STOP")
monitor(result_queue) monitor(result_queue)
+11 -10
View File
@@ -15,6 +15,7 @@ from django.utils import timezone
from django_q.brokers import Broker, get_broker from django_q.brokers import Broker, get_broker
from django_q.cluster import Cluster, Sentinel from django_q.cluster import Cluster, Sentinel
from django_q.conf import Conf from django_q.conf import Conf
from django_q.enums import TimerStatus
from django_q.humanhash import DEFAULT_WORDLIST, uuid from django_q.humanhash import DEFAULT_WORDLIST, uuid
from django_q.models import Success, Task from django_q.models import Success, Task
from django_q.monitor import monitor, save_task from django_q.monitor import monitor, save_task
@@ -245,7 +246,7 @@ def test_cluster(broker):
assert queue_size(broker=broker) == 0 assert queue_size(broker=broker) == 0
# Test work # Test work
task_queue.put("STOP") task_queue.put("STOP")
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
assert task_queue.qsize() == 0 assert task_queue.qsize() == 0
assert result_queue.qsize() == 1 assert result_queue.qsize() == 1
# Test monitor # Test monitor
@@ -272,7 +273,7 @@ def test_results(broker):
pusher(task_queue, stop_event, broker=broker) pusher(task_queue, stop_event, broker=broker)
task_queue.put("STOP") task_queue.put("STOP")
result_queue = Queue() result_queue = Queue()
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
result_queue.put("STOP") result_queue.put("STOP")
monitor(result_queue) monitor(result_queue)
@@ -372,7 +373,7 @@ def test_enqueue(broker, admin_user):
assert fetch_group("test_j", count=2, wait=10) is None assert fetch_group("test_j", count=2, wait=10) is None
# let a worker handle them # let a worker handle them
result_queue = Queue() result_queue = Queue()
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
assert result_queue.qsize() == task_count assert result_queue.qsize() == task_count
result_queue.put("STOP") result_queue.put("STOP")
# store the results # store the results
@@ -541,7 +542,7 @@ def test_recycle(broker, monkeypatch, django_assert_num_queries):
pusher(task_queue, stop_event, broker=broker) pusher(task_queue, stop_event, broker=broker)
pusher(task_queue, stop_event, broker=broker) pusher(task_queue, stop_event, broker=broker)
# worker should exit on recycle # worker should exit on recycle
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
# check if the work has been done # check if the work has been done
assert result_queue.qsize() == 2 assert result_queue.qsize() == 2
# save_limit test # save_limit test
@@ -573,7 +574,7 @@ def test_save_limit_per_func(broker, monkeypatch):
threading.Timer(3, stop_event.set).start() threading.Timer(3, stop_event.set).start()
for i in range(3): for i in range(3):
pusher(task_queue, stop_event, broker=broker) pusher(task_queue, stop_event, broker=broker)
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker) s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker)
assert start_event.is_set() assert start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
@@ -618,7 +619,7 @@ def test_max_rss(broker, monkeypatch):
# push the task # push the task
pusher(task_queue, stop_event, broker=broker) pusher(task_queue, stop_event, broker=broker)
# worker should exit on recycle # worker should exit on recycle
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
# check if the work has been done # check if the work has been done
assert result_queue.qsize() == 1 assert result_queue.qsize() == 1
# save_limit test # save_limit test
@@ -654,7 +655,7 @@ def test_bad_secret(broker, monkeypatch):
worker( worker(
task_queue, task_queue,
result_queue, result_queue,
Value("f", -1), Value("f", TimerStatus.IDLE),
) )
assert result_queue.qsize() == 0 assert result_queue.qsize() == 0
broker.delete_queue() broker.delete_queue()
@@ -838,7 +839,7 @@ class TestSignals:
event.set() event.set()
pusher(task_queue, event, broker=broker) pusher(task_queue, event, broker=broker)
task_queue.put("STOP") task_queue.put("STOP")
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
result_queue.put("STOP") result_queue.put("STOP")
monitor(result_queue, broker) monitor(result_queue, broker)
broker.delete_queue() broker.delete_queue()
@@ -867,7 +868,7 @@ class TestSignals:
event.set() event.set()
pusher(task_queue, event, broker=broker) pusher(task_queue, event, broker=broker)
task_queue.put("STOP") task_queue.put("STOP")
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
result_queue.put("STOP") result_queue.put("STOP")
monitor(result_queue, broker) monitor(result_queue, broker)
broker.delete_queue() broker.delete_queue()
@@ -896,7 +897,7 @@ class TestSignals:
event.set() event.set()
pusher(task_queue, event, broker=broker) pusher(task_queue, event, broker=broker)
task_queue.put("STOP") task_queue.put("STOP")
worker(task_queue, result_queue, Value("f", -1)) worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
result_queue.put("STOP") result_queue.put("STOP")
monitor(result_queue, broker) monitor(result_queue, broker)
broker.delete_queue() broker.delete_queue()
+2 -1
View File
@@ -13,6 +13,7 @@ from django.utils.timezone import is_naive
from django_q.brokers import Broker, get_broker from django_q.brokers import Broker, get_broker
from django_q.conf import Conf from django_q.conf import Conf
from django_q.enums import TimerStatus
from django_q.monitor import monitor from django_q.monitor import monitor
from django_q.pusher import pusher from django_q.pusher import pusher
from django_q.queues import Queue from django_q.queues import Queue
@@ -222,7 +223,7 @@ def test_scheduler(broker, monkeypatch):
task_queue.put("STOP") task_queue.put("STOP")
# let a worker handle them # let a worker handle them
result_queue = Queue() result_queue = Queue()
worker(task_queue, result_queue, Value("b", -1)) worker(task_queue, result_queue, Value("b", TimerStatus.IDLE))
assert result_queue.qsize() == 1 assert result_queue.qsize() == 1
result_queue.put("STOP") result_queue.put("STOP")
# store the results # store the results
+7 -6
View File
@@ -17,6 +17,7 @@ except core.exceptions.AppRegistryNotReady:
django.setup() django.setup()
from django_q.conf import Conf, error_reporter, logger, resource, setproctitle 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.exceptions import TimeoutException
from django_q.signals import post_execute_in_worker, post_spawn, pre_execute from django_q.signals import post_execute_in_worker, post_spawn, pre_execute
from django_q.timeout import TimeoutHandler from django_q.timeout import TimeoutHandler
@@ -54,11 +55,11 @@ def worker(
setproctitle.setproctitle(f"qcluster {proc_name} idle") setproctitle.setproctitle(f"qcluster {proc_name} idle")
task_count = 0 task_count = 0
if timeout is None: if timeout is None:
timeout = -1 timeout = TimerStatus.IDLE
# Start reading the task queue # Start reading the task queue
for task in iter(task_queue.get, "STOP"): for task in iter(task_queue.get, "STOP"):
result = None result = None
timer.value = -1 # Idle timer.value = TimerStatus.IDLE
task_count += 1 task_count += 1
f = task["func"] f = task["func"]
@@ -92,7 +93,7 @@ def worker(
pre_execute.send(sender="django_q", func=f, task=task) pre_execute.send(sender="django_q", func=f, task=task)
# execute the payload # execute the payload
timer.value = timer_value # Busy timer.value = timer_value # Busy
if timer.value != -1: if timer.value != TimerStatus.IDLE:
timer.value += 3 # Add buffer so that guard doesn't kill the process on timeout before it gets processed timer.value += 3 # Add buffer so that guard doesn't kill the process on timeout before it gets processed
timeout_error = False timeout_error = False
@@ -122,15 +123,15 @@ def worker(
result_queue.put(task) result_queue.put(task)
if timeout_error: if timeout_error:
# force destroy process due to timeout # force destroy process due to timeout
timer.value = 0 timer.value = TimerStatus.TIMEOUT
break break
timer.value = -1 # Idle timer.value = TimerStatus.IDLE
if setproctitle: if setproctitle:
setproctitle.setproctitle(f"qcluster {proc_name} idle") setproctitle.setproctitle(f"qcluster {proc_name} idle")
# Recycle # Recycle
if task_count == Conf.RECYCLE or rss_check(): if task_count == Conf.RECYCLE or rss_check():
timer.value = -2 # Recycled timer.value = TimerStatus.RECYCLED
break break
logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name}) logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name})
+1 -1
View File
@@ -27,7 +27,7 @@ Features
- Rollbar and Sentry support - Rollbar and Sentry support
Django Q2 is tested with: Python 3.8, 3.9, 3.10, 3.11 and 3.12. Works with Django 4.2.x, 5.x and 6.0.x Django Q2 is tested with: python 3.10, 3.11, 3.12, 3.13 and 3.14. Works with Django 5.2, 6.0 and 6.1
Currently available in English, German and French. Currently available in English, German and French.