Release v1.7.6

- avoid extra queries when checking the count of success in db

.github/workflows/test.yml

@@ -73,7 +73,7 @@ jobs:
         - name: Install dependencies with Django ${{ matrix.django }}
           run: |
             python -m pip install --upgrade pip
-            pip install poetry
+            pip install poetry==1.8.5
             poetry add "django==${{ matrix.django }}" --python=${{ matrix.python-version }}
             poetry install -E testing
         - name: Run Tests

CHANGELOG.md

@@ -1,5 +1,29 @@
 # Changelog
 
+## [v1.7.6](https://github.com/django-q2/django-q2/tree/v1.7.6) (2025-01-12)
+
+- Make poetry version fixed in CI https://github.com/django-q2/django-q2/pull/260
+
+## [v1.7.5](https://github.com/django-q2/django-q2/tree/v1.7.5) (2025-01-12)
+
+- perf: avoid checking success tasks when save limit is disabled https://github.com/django-q2/django-q2/pull/255
+- Fix install path for CHANGELOG.md https://github.com/django-q2/django-q2/pull/258
+
+## [v1.7.4](https://github.com/django-q2/django-q2/tree/v1.7.4) (2024-11-03)
+
+- Decrease the MAX_RSS set in test_cluster::test_max_rss https://github.com/django-q2/django-q2/pull/240
+- Fix BROKER_CLASS monkeypatch in test_brokers https://github.com/django-q2/django-q2/pull/239
+- Fix 'receive_message_wait_time_seconds' SQS broker management https://github.com/django-q2/django-q2/pull/243
+
+## [v1.7.3](https://github.com/django-q2/django-q2/tree/v1.7.3) (2024-10-15)
+
+- Catch missing SEGALRM with AttributeError instead of ValueError https://github.com/django-q2/django-q2/pull/223
+- Refactor timeout handling to handle AttributeError and ValueError for Windows users https://github.com/django-q2/django-q2/pull/234
+- Fix type check for args in scheduler.py E721 https://github.com/django-q2/django-q2/pull/233
+- Only trigger prometheus if configured https://github.com/django-q2/django-q2/pull/231
+- Fix missing ack_id when finishing task https://github.com/django-q2/django-q2/pull/224
+
+
 ## [v1.7.2](https://github.com/django-q2/django-q2/tree/v1.7.2) (2024-09-09)
 
 - Fix twine check

Dockerfile.dev

@@ -7,6 +7,10 @@ ENV PYTHONUNBUFFERED 1
 # Sets the default shell to bash
 ENV SHELL /bin/bash
 
+RUN set -ex \
+    && apt update \
+    && apt-get install gcc python3-dev --yes
+
 # Upgrades pip
 RUN pip install -U pip setuptools
 

Makefile

@@ -2,16 +2,16 @@ dev:
 	docker compose -f web-docker-compose.yaml up
 
 test:
-	docker-compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run pytest
+	docker compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run pytest
 
 shell:
-	docker-compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run python manage.py shell
+	docker compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run python manage.py shell
 
 makemigrations:
-	docker-compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run python manage.py makemigrations
+	docker compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run python manage.py makemigrations
 
 migrate:
-	docker-compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run python manage.py migrate
+	docker compose -f test-services-docker-compose.yaml run --rm django-q2 poetry run python manage.py migrate
 
 createsuperuser:
 	docker compose -f web-docker-compose.yaml run --rm web python manage.py createsuperuser

containers/localstack/init.sh

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

django_q/__init__.py

@@ -1,6 +1,6 @@
 import django
 
-VERSION = (1, 7, 2)
+VERSION = (1, 7, 6)
 
 if django.VERSION < (3, 2):
     default_app_config = "django_q.apps.DjangoQConfig"

django_q/brokers/aws_sqs.py

@@ -1,3 +1,5 @@
+import copy
+
 from boto3 import Session
 from botocore.client import ClientError
 
@@ -78,15 +80,15 @@ class Sqs(Broker):
 
     @staticmethod
     def get_connection(list_key: str = None) -> Session:
-        config = Conf.SQS
-        if "aws_region" in config:
-            config["region_name"] = config["aws_region"]
-            del config["aws_region"]
+        config_cloned = copy.deepcopy(Conf.SQS)
+        if "aws_region" in config_cloned:
+            config_cloned["region_name"] = config_cloned["aws_region"]
+            del config_cloned["aws_region"]
 
-        if "receive_message_wait_time_seconds" in config:
-            del config["receive_message_wait_time_seconds"]
+        if "receive_message_wait_time_seconds" in config_cloned:
+            del config_cloned["receive_message_wait_time_seconds"]
 
-        return Session(**config)
+        return Session(**config_cloned)
 
     def get_queue(self):
         self.sqs = self.connection.resource("sqs")

django_q/cluster.py

@@ -1,4 +1,5 @@
 # Standard
+import os
 import signal
 import socket
 import uuid
@@ -230,8 +231,14 @@ class Sentinel:
                 % {"name": process.name}
             )
         else:
-            if prometheus_multiprocess:
+            # check if prometheus is proper configurated
+            prometheus_path = os.getenv(
+                "PROMETHEUS_MULTIPROC_DIR", os.getenv("prometheus_multiproc_dir")
+            )
+
+            if prometheus_multiprocess and prometheus_path:
                 prometheus_multiprocess.mark_process_dead(process.pid)
+
             self.pool.remove(process)
             self.spawn_worker()
             if process.timer.value == 0:

django_q/monitor.py

@@ -107,13 +107,16 @@ def save_task(task, broker: Broker):
                 value = get_func_repr(value)
             filters[Conf.SAVE_LIMIT_PER] = value
 
-        with db.transaction.atomic(using=db.router.db_for_write(Success)):
-            list(Success.objects.filter(**filters).select_for_update())
-            if (
-                task["success"]
-                and 0 < Conf.SAVE_LIMIT <= Success.objects.filter(**filters).count()
-            ):
-                Success.objects.filter(**filters).last().delete()
+        # check if we should clean the success tasks
+        if Conf.SAVE_LIMIT > 0:
+            with db.transaction.atomic(using=db.router.db_for_write(Success)):
+                success_tasks_qs = Success.objects.filter(**filters)
+                success_tasks_pks = [
+                    success_task.pk
+                    for success_task in success_tasks_qs.select_for_update()
+                ]
+                if task["success"] and len(success_tasks_pks) >= Conf.SAVE_LIMIT:
+                    success_tasks_qs.last().delete()
 
         # check if this task has previous results
         try:
@@ -145,7 +148,11 @@ def save_task(task, broker: Broker):
                 attempt_count=1,
             )
 
-        if Conf.MAX_ATTEMPTS > 0 and task_obj.attempt_count >= Conf.MAX_ATTEMPTS:
+        if (
+            Conf.MAX_ATTEMPTS > 0
+            and task_obj.attempt_count >= Conf.MAX_ATTEMPTS
+            and task.get("ack_id")
+        ):
             broker.acknowledge(task["ack_id"])
 
     except Exception:

django_q/scheduler.py

@@ -65,7 +65,7 @@ def scheduler(broker: Broker = None):
                 if s.args:
                     args = ast.literal_eval(s.args)
                     # single value won't eval to tuple, so:
-                    if type(args) != tuple:
+                    if type(args) is not tuple:
                         args = (args,)
                 q_options = kwargs.get("q_options", {})
                 if s.intended_date_kwarg:

django_q/tests/test_brokers.py

@@ -51,7 +51,7 @@ def test_redis(monkeypatch):
 
 
 def test_custom(monkeypatch):
-    monkeypatch.setattr(Conf, "BROKER_CLASS", "brokers.redis_broker.Redis")
+    monkeypatch.setattr(Conf, "BROKER_CLASS", "django_q.brokers.redis_broker.Redis")
     broker = get_broker()
     assert broker.ping() is True
     assert broker.info() is not None
@@ -124,7 +124,7 @@ def test_ironmq(monkeypatch):
 @pytest.mark.skipif(
     not os.getenv("AWS_ACCESS_KEY_ID"), reason="requires AWS credentials"
 )
-def canceled_sqs(monkeypatch):
+def test_sqs(monkeypatch):
     monkeypatch.setattr(
         Conf,
         "SQS",
@@ -132,11 +132,13 @@ def canceled_sqs(monkeypatch):
             "aws_region": os.getenv("AWS_REGION"),
             "aws_access_key_id": os.getenv("AWS_ACCESS_KEY_ID"),
             "aws_secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY"),
-            "receive_message_wait_time_seconds": 20,
+            "receive_message_wait_time_seconds": 5,
         },
     )
     # check broker
-    broker = get_broker(list_key=uuid()[0])
+    broker = get_broker(list_key="testing")
+    assert "receive_message_wait_time_seconds" in Conf.SQS
+    assert "aws_region" in Conf.SQS
     assert broker.ping() is True
     assert broker.info() is not None
     assert broker.queue_size() == 0
@@ -173,7 +175,7 @@ def canceled_sqs(monkeypatch):
     broker.enqueue("test")
     while task is None:
         task = broker.dequeue()[0]
-    broker.fail(task[0])
+    broker.fail(task[0][0])
     # bulk test
     for _ in range(10):
         broker.enqueue("test")

django_q/tests/test_cluster.py

@@ -399,7 +399,7 @@ def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs
 
 
 @pytest.mark.django_db
-def test_recycle(broker, monkeypatch):
+def test_recycle(broker, monkeypatch, django_assert_num_queries):
     # set up the Sentinel
     broker.list_key = "test_recycle_test:q"
     async_task("django_q.tests.tasks.multiply", 2, 2, broker=broker)
@@ -432,7 +432,8 @@ def test_recycle(broker, monkeypatch):
     monkeypatch.setattr(Conf, "SAVE_LIMIT", 1)
     result_queue.put("STOP")
     # run monitor
-    monitor(result_queue)
+    with django_assert_num_queries(12):
+        monitor(result_queue)
     assert Success.objects.count() == Conf.SAVE_LIMIT
     broker.delete_queue()
 
@@ -487,7 +488,7 @@ def test_max_rss(broker, monkeypatch):
     stop_event = Event()
     cluster_id = uuidlib.uuid4()
     # override settings
-    monkeypatch.setattr(Conf, "MAX_RSS", 40000)
+    monkeypatch.setattr(Conf, "MAX_RSS", 20000)
     monkeypatch.setattr(Conf, "WORKERS", 1)
     # set a timer to stop the Sentinel
     threading.Timer(3, stop_event.set).start()

django_q/timeout.py

@@ -22,11 +22,13 @@ class TimeoutHandler:
             return
         try:
             signal.signal(signal.SIGALRM, self.raise_timeout_exception)
-        except ValueError:  # ValueError is raised for Windows users
+            signal.alarm(self._timeout)
+        except (
+            ValueError,
+            AttributeError,
+        ):  # AttributeError or ValueError might be raised for Windows users
             logger.debug(_("SIGALARM is not available on your platform"))
 
-        signal.alarm(self._timeout)
-
     def __exit__(self, exc_type, exc_value, traceback):
         if self._timeout == -1:
             return
@@ -34,5 +36,8 @@ class TimeoutHandler:
         try:
             signal.alarm(0)
             signal.signal(signal.SIGALRM, signal.SIG_DFL)
-        except ValueError:  # ValueError is raised for Windows users
+        except (
+            ValueError,
+            AttributeError,
+        ):  # AttributeError or ValueError might be raised for Windows users
             logger.debug(_("SIGALARM is not available on your platform"))

docs/conf.py

@@ -75,7 +75,7 @@ author = "Ilan Steemers, Stan Triepels"
 # The short X.Y version.
 version = "1.7"
 # The full version, including alpha/beta/rc tags.
-release = "1.7.2"
+release = "1.7.6"
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "django-q2"
-version = "1.7.2"
+version = "1.7.6"
 packages = [
     { include = "django_q" },
 ]
@@ -39,7 +39,10 @@ classifiers = [
     'Topic :: System :: Distributed Computing',
     'Topic :: Software Development :: Libraries :: Python Modules',
 ]
-include = ['CHANGELOG.md']
+
+include = [
+ { path = "CHANGELOG.md", format = "sdist" },
+]
 
 [tool.poetry.plugins] # Optional super table
 

test-services-docker-compose.yaml

@@ -13,15 +13,41 @@ services:
     networks:
       - main
 
+  aws:
+    container_name: aws
+    image: localstack/localstack:3.4.0
+    ports:
+      - "127.0.0.1:4566:4566"            # LocalStack Gateway
+      - "127.0.0.1:4510-4559:4510-4559"  # External services port range
+    environment:
+      AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
+      DEFAULT_REGION: ${AWS_DEFAULT_REGION:-us-west-2}
+      SQS_ENDPOINT_STRATEGY: path
+      SERVICES: sqs
+      LOCALSTACK_HOST: aws
+      DEBUG: 1
+      LS_LOG: trace
+    volumes:
+      - ./containers/localstack:/etc/localstack/init/ready.d
+    networks:
+      - main
+
   django-q2:
     build:
       dockerfile: ./Dockerfile.dev
       context: .
+    environment:
+      AWS_ENDPOINT_URL: http://aws:4566
+      AWS_REGION: ${AWS_REGION:-us-west-2}
+      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}
     volumes:
       - .:/app
     depends_on:
       - redis
       - mongo
+      - aws
     networks:
       - main