Merge branch 'main' into feature/support_rediscluster_transport

Author: auvipy

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 5.5.4
+current_version = 5.6.0b1
 commit = True
 tag = True
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(?P<releaselevel>[a-z]+)?

Changelog.rst

@@ -4,6 +4,31 @@
  Change history
 ================
 
+.. _version-5.6.0b1:
+
+5.6.0b1
+=======
+:release-date: 29 July, 2025
+:release-by: Tomer Nosrati
+
+What's Changed
+~~~~~~~~~~~~~~
+
+- Revert "Feature: urllib3 instead of curl" (#2261)
+- Fix #2286 : SQS - Enhance support for receiving message attributes. Allow string in msg body. (#2300)
+- Update gcpubsub.txt for protobuf (#2320)
+- fix body bytes encoding in curl.py (#2322)
+- Hotfix(aws_ssl): validate secure connection (#2323)
+- Update setup.py to remove deprecation warning (#2298)
+- Update protobuf to 5.29.5 (#2199)
+- Allow setting `tags` in initial `boto3.sqs.create_queue` call via `transport_options` (#2321)
+- Add sts_token_buffer_time parameter to transport options (#2216)
+- Docs: Fix incorrect "added in 5.6.0" note for sts_token_timeout in SQS transport docs (#2327)
+- SQS: when redelivering a message apply visibility timeout based on wait_time_seconds (long polling time) (#2049)
+- Bump redis requirement to <6 to support v5.3 (#2329)
+- Pidbox: support queue_exclusive / queue_durable (RabbitMQ 4.x) (#2338)
+- Prepare for (pre) release: v5.6.0b1 (#2343)
+
 .. _version-5.5.4:
 
 5.5.4
@@ -23,8 +48,6 @@ What's Changed
 - Fixed lint error from `kombu/transport/native_delayed_delivery.py` (#2308)
 - Prepare for release: v5.5.4 (#2309)
 
-.. _version-5.5.2:
-
 .. _version-5.5.3:
 
 5.5.3

README.rst

@@ -4,7 +4,7 @@
 
 |build-status| |coverage| |license| |wheel| |pyversion| |pyimp| |downloads|
 
-:Version: 5.5.4
+:Version: 5.6.0b1
 :Documentation: https://kombu.readthedocs.io/
 :Download: https://pypi.org/project/kombu/
 :Source: https://github.com/celery/kombu/

docs/includes/introduction.txt

@@ -1,4 +1,4 @@
-:Version: 5.5.4
+:Version: 5.6.0b1
 :Web: https://kombu.readthedocs.io/
 :Download: https://pypi.org/project/kombu/
 :Source: https://github.com/celery/kombu/

docs/reference/kombu.pidbox.rst

@@ -70,6 +70,20 @@
         .. automethod:: get_reply_queue
         .. automethod:: get_queue
 
+    Mailbox Options
+    ~~~~~~~~~~~~~~~
+
+    .. versionadded:: 5.6.0
+
+    The `Mailbox` supports several configuration options that affect
+    the behavior of its exchanges and queues.
+
+    - ``durable``: If True, declares durable exchanges that survive broker restarts.
+    - ``exclusive``: If True, declares exclusive exchanges (usable by only one connection).
+
+    These provide finer control over broker-side behavior and are useful
+    in production environments where queue durability matters.\
+
     Node
     ----
 
@@ -91,4 +105,3 @@
         .. automethod:: handle
         .. automethod:: handle_message
         .. automethod:: reply
-

kombu/__init__.py

@@ -8,7 +8,7 @@
 from collections import namedtuple
 from typing import Any, cast
 
-__version__ = '5.5.4'
+__version__ = '5.6.0b1'
 __author__ = 'Ask Solem'
 __contact__ = '[email protected]'
 __homepage__ = 'https://kombu.readthedocs.io'

kombu/pidbox.py

@@ -181,6 +181,7 @@ def __init__(self, namespace,
                  type='direct', connection=None, clock=None,
                  accept=None, serializer=None, producer_pool=None,
                  queue_ttl=None, queue_expires=None,
+                 queue_durable=False, queue_exclusive=False,
                  reply_queue_ttl=None, reply_queue_expires=10.0):
         self.namespace = namespace
         self.connection = connection
@@ -193,9 +194,16 @@ def __init__(self, namespace,
         self.serializer = self.serializer if serializer is None else serializer
         self.queue_ttl = queue_ttl
         self.queue_expires = queue_expires
+        self.queue_durable = queue_durable
+        self.queue_exclusive = queue_exclusive
         self.reply_queue_ttl = reply_queue_ttl
         self.reply_queue_expires = reply_queue_expires
         self._producer_pool = producer_pool
+        if queue_exclusive and queue_durable:
+            raise ValueError(
+                "queue_exclusive and queue_durable cannot both be True "
+                "(exclusive queues are automatically deleted and cannot be durable).",
+            )
 
     def __call__(self, connection):
         bound = copy(self)
@@ -236,8 +244,9 @@ def get_reply_queue(self):
             f'{oid}.{self.reply_exchange.name}',
             exchange=self.reply_exchange,
             routing_key=oid,
-            durable=False,
-            auto_delete=True,
+            durable=self.queue_durable,
+            exclusive=self.queue_exclusive,
+            auto_delete=not self.queue_durable,
             expires=self.reply_queue_expires,
             message_ttl=self.reply_queue_ttl,
         )
@@ -250,8 +259,9 @@ def get_queue(self, hostname):
         return Queue(
             f'{hostname}.{self.namespace}.pidbox',
             exchange=self.exchange,
-            durable=False,
-            auto_delete=True,
+            durable=self.queue_durable,
+            exclusive=self.queue_exclusive,
+            auto_delete=not self.queue_durable,
             expires=self.queue_expires,
             message_ttl=self.queue_ttl,
         )

requirements/extras/redis.txt

@@ -1 +1 @@
-redis>=4.5.2,!=4.5.5,!=5.0.2,<=5.2.1
+redis>=4.5.2,!=4.5.5,!=5.0.2,<6

t/unit/test_pidbox.py

@@ -342,6 +342,38 @@ def get_next(self, consumer):
         if m:
             return m.payload
 
+    def test_mailbox_queue_exclusive(self):
+        mbox = pidbox.Mailbox(
+            'flagbox_ex',
+            queue_exclusive=True,
+            queue_durable=False,
+        )(self.connection)
+
+        for q in (mbox.get_queue('worker1'), mbox.get_reply_queue()):
+            assert q.exclusive is True
+            assert q.durable is False
+            assert q.auto_delete is True
+
+    def test_mailbox_queue_durable(self):
+        mbox = pidbox.Mailbox(
+            'flagbox_dur',
+            queue_exclusive=False,
+            queue_durable=True,
+        )(self.connection)
+
+        for q in (mbox.get_queue('worker1'), mbox.get_reply_queue()):
+            assert q.durable is True
+            assert q.exclusive is False
+            assert q.auto_delete is False
+
+    def test_mailbox_invalid_flag_combo(self):
+        with pytest.raises(ValueError):
+            pidbox.Mailbox(
+                'flagbox_bad',
+                queue_exclusive=True,
+                queue_durable=True,
+            )(self.connection)
+
 
 GLOBAL_PIDBOX = pidbox.Mailbox('global_unittest_mailbox')