mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-30 10:48:11 +08:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a8f3bfb832 |
+24
-32
@@ -6,21 +6,6 @@ on:
|
|||||||
pull_request:
|
pull_request:
|
||||||
branches:
|
branches:
|
||||||
- master
|
- 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:
|
jobs:
|
||||||
lint:
|
lint:
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
@@ -83,13 +68,6 @@ jobs:
|
|||||||
ports:
|
ports:
|
||||||
- 6379:6379
|
- 6379:6379
|
||||||
options: --entrypoint redis-server
|
options: --entrypoint redis-server
|
||||||
aws:
|
|
||||||
image: ministackorg/ministack:1.4.16
|
|
||||||
env:
|
|
||||||
SERVICES: sqs
|
|
||||||
MINISTACK_HOST: aws
|
|
||||||
ports:
|
|
||||||
- 4566:4566
|
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v3
|
- uses: actions/checkout@v3
|
||||||
- name: Set up Python ${{ matrix.python-version }}
|
- name: Set up Python ${{ matrix.python-version }}
|
||||||
@@ -101,16 +79,30 @@ jobs:
|
|||||||
python -m pip install uv==0.12.1
|
python -m pip install uv==0.12.1
|
||||||
uv sync --extra testing --python ${{ matrix.python-version }}
|
uv sync --extra testing --python ${{ matrix.python-version }}
|
||||||
uv pip install "django~=${{ matrix.django }}a1"
|
uv pip install "django~=${{ matrix.django }}a1"
|
||||||
- name: Create SQS queue
|
|
||||||
run: aws sqs create-queue --queue-name testing
|
|
||||||
- name: Run Tests
|
- name: Run Tests
|
||||||
run: |
|
run: |
|
||||||
uv run pytest --cov=./django_q --cov-report=xml --junitxml=pytest.xml --cov-report=term-missing:skip-covered | tee pytest-coverage.txt
|
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
|
||||||
|
|
||||||
- name: Coverage comment
|
finish:
|
||||||
if: matrix.python-version == '3.13' && matrix.django == '5.2' && github.event_name == 'pull_request'
|
needs: test
|
||||||
uses: MishaKav/pytest-coverage-comment@v1
|
runs-on: ubuntu-latest
|
||||||
with:
|
container: python:3.11-bookworm
|
||||||
pytest-coverage-path: ./pytest-coverage.txt
|
steps:
|
||||||
junitxml-path: ./pytest.xml
|
- name: Upload to coveralls
|
||||||
report-only-changed-files: true
|
run: |
|
||||||
|
python -m pip install --upgrade pip
|
||||||
|
python -m pip install coveralls
|
||||||
|
coveralls --service=github --finish
|
||||||
|
env:
|
||||||
|
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||||
|
|||||||
@@ -1,21 +1,5 @@
|
|||||||
# Changelog
|
# 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)
|
## [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
|
- fix: Fix incorrect signal import (#308) https://github.com/django-q2/django-q2/pull/308
|
||||||
|
|||||||
Executable
+19
@@ -0,0 +1,19 @@
|
|||||||
|
#!/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
|
||||||
@@ -1 +0,0 @@
|
|||||||
aws sqs create-queue --queue-name testing
|
|
||||||
@@ -1,3 +1,3 @@
|
|||||||
VERSION = (1, 11, 1)
|
VERSION = (1, 10, 0)
|
||||||
|
|
||||||
__all__ = ["conf", "cluster", "models", "tasks"]
|
__all__ = ["conf", "cluster", "models", "tasks"]
|
||||||
|
|||||||
+11
-9
@@ -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
|
||||||
@@ -41,10 +42,7 @@ from django_q.worker import worker
|
|||||||
|
|
||||||
|
|
||||||
def get_mp_context():
|
def get_mp_context():
|
||||||
if "fork" in multiprocessing.get_all_start_methods():
|
return multiprocessing.get_context("fork")
|
||||||
return multiprocessing.get_context("fork")
|
|
||||||
else:
|
|
||||||
return multiprocessing.get_context()
|
|
||||||
|
|
||||||
|
|
||||||
class Cluster:
|
class Cluster:
|
||||||
@@ -237,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:
|
||||||
@@ -274,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 = ""
|
||||||
@@ -299,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(
|
||||||
@@ -351,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():
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -172,6 +172,9 @@ def test_ironmq(monkeypatch):
|
|||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
not os.getenv("AWS_ACCESS_KEY_ID"), reason="requires AWS credentials"
|
||||||
|
)
|
||||||
def test_sqs(monkeypatch):
|
def test_sqs(monkeypatch):
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
Conf,
|
Conf,
|
||||||
@@ -185,7 +188,6 @@ def test_sqs(monkeypatch):
|
|||||||
)
|
)
|
||||||
# check broker
|
# check broker
|
||||||
broker = get_broker(list_key="testing")
|
broker = get_broker(list_key="testing")
|
||||||
broker.purge_queue()
|
|
||||||
assert "receive_message_wait_time_seconds" in Conf.SQS
|
assert "receive_message_wait_time_seconds" in Conf.SQS
|
||||||
assert "aws_region" in Conf.SQS
|
assert "aws_region" in Conf.SQS
|
||||||
assert broker.ping() is True
|
assert broker.ping() is True
|
||||||
@@ -222,9 +224,8 @@ def test_sqs(monkeypatch):
|
|||||||
assert broker.dequeue() is None
|
assert broker.dequeue() is None
|
||||||
# fail
|
# fail
|
||||||
broker.enqueue("test")
|
broker.enqueue("test")
|
||||||
task = None
|
|
||||||
while task is None:
|
while task is None:
|
||||||
task = broker.dequeue()
|
task = broker.dequeue()[0]
|
||||||
broker.fail(task[0][0])
|
broker.fail(task[0][0])
|
||||||
# bulk test
|
# bulk test
|
||||||
for _ in range(10):
|
for _ in range(10):
|
||||||
@@ -234,6 +235,8 @@ def test_sqs(monkeypatch):
|
|||||||
for task in tasks:
|
for task in tasks:
|
||||||
assert task is not None
|
assert task is not None
|
||||||
broker.acknowledge(task[0])
|
broker.acknowledge(task[0])
|
||||||
|
# duplicate acknowledge
|
||||||
|
broker.acknowledge(task[0])
|
||||||
assert broker.lock_size() == 0
|
assert broker.lock_size() == 0
|
||||||
# delete queue
|
# delete queue
|
||||||
broker.enqueue("test")
|
broker.enqueue("test")
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -1,4 +1,3 @@
|
|||||||
import multiprocessing
|
|
||||||
import os
|
import os
|
||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
@@ -14,8 +13,9 @@ import pytest
|
|||||||
from django.utils import timezone
|
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, get_mp_context
|
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
|
||||||
@@ -60,53 +60,6 @@ def broker(monkeypatch):
|
|||||||
return get_broker()
|
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):
|
def test_redis_connection(broker):
|
||||||
assert broker.ping() is True
|
assert broker.ping() is True
|
||||||
|
|
||||||
@@ -293,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
|
||||||
@@ -320,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)
|
||||||
|
|
||||||
@@ -420,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
|
||||||
@@ -589,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
|
||||||
@@ -621,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
|
||||||
@@ -666,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
|
||||||
@@ -702,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()
|
||||||
@@ -886,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()
|
||||||
@@ -915,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()
|
||||||
@@ -944,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()
|
||||||
|
|||||||
@@ -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
@@ -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})
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -73,9 +73,9 @@ author = "Ilan Steemers, Stan Triepels"
|
|||||||
# built documents.
|
# built documents.
|
||||||
#
|
#
|
||||||
# The short X.Y version.
|
# The short X.Y version.
|
||||||
version = "1.11"
|
version = "1.10"
|
||||||
# The full version, including alpha/beta/rc tags.
|
# The full version, including alpha/beta/rc tags.
|
||||||
release = "1.11.1"
|
release = "1.10.0"
|
||||||
|
|
||||||
# The language for content autogenerated by Sphinx. Refer to documentation
|
# The language for content autogenerated by Sphinx. Refer to documentation
|
||||||
# for a list of supported languages.
|
# for a list of supported languages.
|
||||||
|
|||||||
+2
-2
@@ -4,7 +4,7 @@ build-backend = "uv_build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "django-q2"
|
name = "django-q2"
|
||||||
version = "1.11.1"
|
version = "1.10.0"
|
||||||
packages = [
|
packages = [
|
||||||
{ include = "django_q" },
|
{ include = "django_q" },
|
||||||
]
|
]
|
||||||
@@ -33,11 +33,11 @@ classifiers = [
|
|||||||
"Operating System :: MacOS",
|
"Operating System :: MacOS",
|
||||||
"Programming Language :: Python",
|
"Programming Language :: Python",
|
||||||
"Programming Language :: Python :: 3",
|
"Programming Language :: Python :: 3",
|
||||||
|
"Programming Language :: Python :: 3.9",
|
||||||
"Programming Language :: Python :: 3.10",
|
"Programming Language :: Python :: 3.10",
|
||||||
"Programming Language :: Python :: 3.11",
|
"Programming Language :: Python :: 3.11",
|
||||||
"Programming Language :: Python :: 3.12",
|
"Programming Language :: Python :: 3.12",
|
||||||
"Programming Language :: Python :: 3.13",
|
"Programming Language :: Python :: 3.13",
|
||||||
"Programming Language :: Python :: 3.14",
|
|
||||||
"Topic :: Internet :: WWW/HTTP",
|
"Topic :: Internet :: WWW/HTTP",
|
||||||
"Topic :: System :: Distributed Computing",
|
"Topic :: System :: Distributed Computing",
|
||||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||||
|
|||||||
@@ -15,18 +15,20 @@ services:
|
|||||||
|
|
||||||
aws:
|
aws:
|
||||||
container_name: aws
|
container_name: aws
|
||||||
image: ministackorg/ministack:1.4.16
|
image: localstack/localstack:3.4.0
|
||||||
ports:
|
ports:
|
||||||
- "127.0.0.1:4566:4566"
|
- "127.0.0.1:4566:4566" # LocalStack Gateway
|
||||||
|
- "127.0.0.1:4510-4559:4510-4559" # External services port range
|
||||||
environment:
|
environment:
|
||||||
MINISTACK_HOST: aws
|
AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
|
||||||
|
DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
|
||||||
|
SQS_ENDPOINT_STRATEGY: path
|
||||||
SERVICES: sqs
|
SERVICES: sqs
|
||||||
|
LOCALSTACK_HOST: aws
|
||||||
|
DEBUG: 1
|
||||||
|
LS_LOG: trace
|
||||||
volumes:
|
volumes:
|
||||||
- ./containers/ministack:/docker-entrypoint-initaws.d
|
- ./containers/localstack:/etc/localstack/init/ready.d
|
||||||
healthcheck:
|
|
||||||
test: ["CMD", "wget", "-qO-", "http://localhost:4566/_ministack/ready"]
|
|
||||||
interval: 2s
|
|
||||||
retries: 15
|
|
||||||
networks:
|
networks:
|
||||||
- main
|
- main
|
||||||
|
|
||||||
@@ -36,10 +38,10 @@ services:
|
|||||||
context: .
|
context: .
|
||||||
environment:
|
environment:
|
||||||
AWS_ENDPOINT_URL: http://aws:4566
|
AWS_ENDPOINT_URL: http://aws:4566
|
||||||
AWS_REGION: ${AWS_REGION:-us-east-1}
|
AWS_REGION: ${AWS_REGION:-us-west-2}
|
||||||
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID:-test}
|
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID:-test}
|
||||||
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY:-test}
|
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY:-test}
|
||||||
AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-east-1}
|
AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
|
||||||
volumes:
|
volumes:
|
||||||
- .:/app
|
- .:/app
|
||||||
depends_on:
|
depends_on:
|
||||||
|
|||||||
Reference in New Issue
Block a user