5 Commits

Author SHA1 Message Date
Stan
b0c6753462 Release v1.11.1 2026-08-26 04:46:29 +02:00
David Macario
d97b27901b Fix cluster requesting hardcoded unix-only fork context (#347) 2026-08-26 04:39:33 +02:00
Stan Triepels
fa935f89ab Show test coverage in github pull request comment (#344) 2026-08-14 17:26:55 +02:00
Stan Triepels
9e576b85e6 Swap localstorage to ministack (#343) 2026-08-14 05:37:10 +02:00
Stan
1941f1e3f0 Release v1.11.0 2026-08-10 04:17:05 +02:00
15 changed files with 143 additions and 114 deletions

View File

@@ -6,6 +6,21 @@ on:
pull_request:
branches:
- master
# for code coverage comment
permissions:
contents: read
pull-requests: write
env:
MONGO_HOST: "127.0.0.1"
REDIS_HOST: "127.0.0.1"
AWS_ENDPOINT_URL: http://localhost:4566
AWS_REGION: "us-east-1"
AWS_ACCESS_KEY_ID: "test"
AWS_SECRET_ACCESS_KEY: "test"
AWS_DEFAULT_REGION: "us-east-1"
jobs:
lint:
runs-on: ubuntu-latest
@@ -68,6 +83,13 @@ jobs:
ports:
- 6379:6379
options: --entrypoint redis-server
aws:
image: ministackorg/ministack:1.4.16
env:
SERVICES: sqs
MINISTACK_HOST: aws
ports:
- 4566:4566
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
@@ -79,30 +101,16 @@ jobs:
python -m pip install uv==0.12.1
uv sync --extra testing --python ${{ matrix.python-version }}
uv pip install "django~=${{ matrix.django }}a1"
- name: Create SQS queue
run: aws sqs create-queue --queue-name testing
- name: Run Tests
run: |
uv run pytest --cov=./django_q --cov-report=xml
env:
MONGO_HOST: "127.0.0.1"
REDIS_HOST: "127.0.0.1"
- name: Upload to coveralls
run: |
python -m pip install coveralls
coveralls --service=github
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
COVERALLS_FLAG_NAME: python-${{ matrix.python-version }}-django-${{ matrix.django }}
COVERALLS_PARALLEL: true
uv run pytest --cov=./django_q --cov-report=xml --junitxml=pytest.xml --cov-report=term-missing:skip-covered | tee pytest-coverage.txt
finish:
needs: test
runs-on: ubuntu-latest
container: python:3.11-bookworm
steps:
- name: Upload to coveralls
run: |
python -m pip install --upgrade pip
python -m pip install coveralls
coveralls --service=github --finish
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Coverage comment
if: matrix.python-version == '3.13' && matrix.django == '5.2' && github.event_name == 'pull_request'
uses: MishaKav/pytest-coverage-comment@v1
with:
pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml
report-only-changed-files: true

View File

@@ -1,5 +1,21 @@
# Changelog
## [v1.11.1](https://github.com/django-q2/django-q2/tree/v1.11.1) (2026-08-26)
- Swap localstorage to ministack (#343) https://github.com/django-q2/django-q2/pull/343
- Show test coverage in github pull request comment (#344) https://github.com/django-q2/django-q2/pull/344
- Fix cluster requesting hardcoded unix-only fork context (#347) https://github.com/django-q2/django-q2/pull/347
## [v1.11.0](https://github.com/django-q2/django-q2/tree/v1.11.0) (2026-08-10)
- AttributeError when start_event is None, and guard process faster stop (#305) https://github.com/django-q2/django-q2/pull/305
- Add croniter as optional extra. (#336) https://github.com/django-q2/django-q2/pull/336
- Migrate poetry to uv (#337) https://github.com/django-q2/django-q2/pull/337
- Add support for python 3.13 and 3.14 (#338) https://github.com/django-q2/django-q2/pull/338
- Add django 6.1 support (#340) https://github.com/django-q2/django-q2/pull/340
- Postpone SECRET_KEY evaluation to the qcluster command execution (#332) https://github.com/django-q2/django-q2/pull/332
## [v1.10.0](https://github.com/django-q2/django-q2/tree/v1.10.0) (2026-05-01)
- fix: Fix incorrect signal import (#308) https://github.com/django-q2/django-q2/pull/308

View File

@@ -1,19 +0,0 @@
#!/bin/bash
# Note that this file needs to have the executable bit set for it to work with later localstack implementations.
export DEFAULT_REGION=us-west-2
create_sqs() {
QUEUE_NAME="$1"
TIMEOUT=${2:-60}
DL_QUEUE_URL=$(awslocal sqs create-queue --queue-name "dl-$QUEUE_NAME" --query QueueUrl --output text)
echo ">>> Created $DL_QUEUE_URL queue!"
DL_QUEUE_ARN=$(awslocal sqs get-queue-attributes --queue-url "$DL_QUEUE_URL" --attribute-names QueueArn --query Attributes.QueueArn --output text)
awslocal sqs create-queue --queue-name "$QUEUE_NAME" --attributes '{
"RedrivePolicy": "{\"deadLetterTargetArn\": \"'"$DL_QUEUE_ARN"'\",\"maxReceiveCount\":\"3\"}",
"VisibilityTimeout": "'"$TIMEOUT"'"
}'
}
# Create SQS queues
create_sqs testing

View File

@@ -0,0 +1 @@
aws sqs create-queue --queue-name testing

View File

@@ -1,3 +1,3 @@
VERSION = (1, 10, 0)
VERSION = (1, 11, 1)
__all__ = ["conf", "cluster", "models", "tasks"]

View File

@@ -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
@@ -42,7 +41,10 @@ from django_q.worker import worker
def get_mp_context():
return multiprocessing.get_context("fork")
if "fork" in multiprocessing.get_all_start_methods():
return multiprocessing.get_context("fork")
else:
return multiprocessing.get_context()
class Cluster:
@@ -235,11 +237,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 +274,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 +299,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 +351,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():

View File

@@ -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

View File

@@ -172,9 +172,6 @@ def test_ironmq(monkeypatch):
broker.delete_queue()
@pytest.mark.skipif(
not os.getenv("AWS_ACCESS_KEY_ID"), reason="requires AWS credentials"
)
def test_sqs(monkeypatch):
monkeypatch.setattr(
Conf,
@@ -188,6 +185,7 @@ def test_sqs(monkeypatch):
)
# check broker
broker = get_broker(list_key="testing")
broker.purge_queue()
assert "receive_message_wait_time_seconds" in Conf.SQS
assert "aws_region" in Conf.SQS
assert broker.ping() is True
@@ -224,8 +222,9 @@ def test_sqs(monkeypatch):
assert broker.dequeue() is None
# fail
broker.enqueue("test")
task = None
while task is None:
task = broker.dequeue()[0]
task = broker.dequeue()
broker.fail(task[0][0])
# bulk test
for _ in range(10):
@@ -235,8 +234,6 @@ def test_sqs(monkeypatch):
for task in tasks:
assert task is not None
broker.acknowledge(task[0])
# duplicate acknowledge
broker.acknowledge(task[0])
assert broker.lock_size() == 0
# delete queue
broker.enqueue("test")

View File

@@ -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)

View File

@@ -1,3 +1,4 @@
import multiprocessing
import os
import signal
import sys
@@ -13,9 +14,8 @@ import pytest
from django.utils import timezone
from django_q.brokers import Broker, get_broker
from django_q.cluster import Cluster, Sentinel
from django_q.cluster import Cluster, Sentinel, get_mp_context
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
@@ -60,6 +60,53 @@ def broker(monkeypatch):
return get_broker()
def test_get_mp_context_prefers_fork_when_available(monkeypatch):
monkeypatch.setattr(
multiprocessing,
"get_all_start_methods",
lambda: ["fork", "spawn", "forkserver"],
)
calls = []
class DummyContext:
def get_start_method(self):
return "fork"
def fake_get_context(method=None):
calls.append(method)
return DummyContext()
monkeypatch.setattr(multiprocessing, "get_context", fake_get_context)
assert get_mp_context().get_start_method() == "fork"
assert calls == ["fork"]
def test_get_mp_context_falls_back_to_platform_default_without_fork(monkeypatch):
"""
Regression test: get_mp_context() used to hardcode the unix-only "fork"
context, which raises ValueError on platforms (e.g. Windows) that don't
support it. It should instead defer to the platform's default context
whenever "fork" isn't available.
"""
monkeypatch.setattr(multiprocessing, "get_all_start_methods", lambda: ["spawn"])
calls = []
real_get_context = multiprocessing.get_context
def fake_get_context(method=None):
calls.append(method)
return real_get_context(method)
monkeypatch.setattr(multiprocessing, "get_context", fake_get_context)
get_mp_context()
# Must ask for the platform default (no explicit method), should never be equal to "fork"
assert calls == [None]
def test_redis_connection(broker):
assert broker.ping() is True
@@ -246,7 +293,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 +320,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 +420,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 +589,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 +621,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 +666,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 +702,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 +886,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 +915,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 +944,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()

View File

@@ -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

View File

@@ -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})

View File

@@ -73,9 +73,9 @@ author = "Ilan Steemers, Stan Triepels"
# built documents.
#
# The short X.Y version.
version = "1.10"
version = "1.11"
# The full version, including alpha/beta/rc tags.
release = "1.10.0"
release = "1.11.1"
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.

View File

@@ -4,7 +4,7 @@ build-backend = "uv_build"
[project]
name = "django-q2"
version = "1.10.0"
version = "1.11.1"
packages = [
{ include = "django_q" },
]
@@ -33,11 +33,11 @@ classifiers = [
"Operating System :: MacOS",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Topic :: Internet :: WWW/HTTP",
"Topic :: System :: Distributed Computing",
"Topic :: Software Development :: Libraries :: Python Modules",

View File

@@ -15,20 +15,18 @@ services:
aws:
container_name: aws
image: localstack/localstack:3.4.0
image: ministackorg/ministack:1.4.16
ports:
- "127.0.0.1:4566:4566" # LocalStack Gateway
- "127.0.0.1:4510-4559:4510-4559" # External services port range
- "127.0.0.1:4566:4566"
environment:
AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
SQS_ENDPOINT_STRATEGY: path
MINISTACK_HOST: aws
SERVICES: sqs
LOCALSTACK_HOST: aws
DEBUG: 1
LS_LOG: trace
volumes:
- ./containers/localstack:/etc/localstack/init/ready.d
- ./containers/ministack:/docker-entrypoint-initaws.d
healthcheck:
test: ["CMD", "wget", "-qO-", "http://localhost:4566/_ministack/ready"]
interval: 2s
retries: 15
networks:
- main
@@ -38,10 +36,10 @@ services:
context: .
environment:
AWS_ENDPOINT_URL: http://aws:4566
AWS_REGION: ${AWS_REGION:-us-west-2}
AWS_REGION: ${AWS_REGION:-us-east-1}
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID:-test}
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY:-test}
AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-east-1}
volumes:
- .:/app
depends_on: