feat: do not create delay queue by default

Author: danfimov

.pre-commit-config.yaml

@@ -1,10 +1,6 @@
-# See https://pre-commit.com for more information
-# See https://pre-commit.com/hooks.html for more hooks
 repos:
-  - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v6.0.0
+  - repo: builtin
     hooks:
-      - id: check-ast
       - id: trailing-whitespace
       - id: check-toml
       - id: end-of-file-fixer
@@ -47,12 +43,13 @@ repos:
           - "taskiq_aio_pika"
           - "tests"
 
-      - id: mypy
-        name: Validate types with MyPy
-        entry: uv run mypy
+      - id: ty
+        name: Validate types with ty
+        entry: uv run ty
         language: system
         pass_filenames: false
         types: [python]
         args:
+          - "check"
           - ./taskiq_aio_pika
           - ./tests

README.md

@@ -28,7 +28,18 @@ async def test() -> None:
 
 ### Default delays
 
-To send delayed message, you have to specify delay label. You can do it with `task` decorator, or by using kicker.
+To send delayed messagem, you need to specify queue for delayed messages. You can do it by passing `delay_queue` parameter to the broker. For example:
+
+```python
+from taskiq_aio_pika import AioPikaBroker, Queue, QueueType
+
+broker = AioPikaBroker(
+    ...,
+    delay_queue=Queue(name="taskiq.delay_queue"),
+)
+```
+
+After that you have to specify delay label. You can do it with `task` decorator, or by using kicker.
 
 In this type of delay we are using additional queue with `expiration` parameter. After declared time message will be deleted from `delay` queue and sent to the main queue. For example:
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "taskiq-aio-pika"
-version = "0.0.0"
+version = "0.6.0"
 description = "RabbitMQ broker for taskiq"
 authors = [
     {name = "Pavel Kirilin", email = "win10@list.ru"}
@@ -37,36 +37,30 @@ dependencies = [
 
 [dependency-groups]
 dev = [
-    "pre-commit>=4.4.0",
-    # lint
+    "prek>=0.3.0",
+    {include-group = "test"},
+    {include-group = "lint"},
+    {include-group = "typecheck"},
+    {include-group = "examples"},
+]
+lint = [
     "ruff>=0.14.5",
     "black>=25.11.0",
-    # type check
-    "mypy>=1.18.2",
-    # tests
+]
+test = [
     "pytest>=9.0.1",
     "pytest-cov>=7.0.0",
     "coverage>=7.11.3",
     "pytest-xdist[psutil]>=3.8.0",
     "anyio>=4.11.0",
-    {include-group = "examples"},
+]
+typecheck = [
+    "ty>=0.0.13",
 ]
 examples = [
     "taskiq-redis>=1.1.2",
 ]
 
-[tool.mypy]
-python_version = "3.10"
-strict = true
-ignore_missing_imports = true
-allow_subclassing_any = true
-allow_untyped_calls = true
-pretty = true
-show_error_codes = true
-implicit_reexport = true
-allow_untyped_decorators = true
-warn_return_any = false
-
 [build-system]
 requires = ["uv_build>=0.9.9,<0.10.0"]
 build-backend = "uv_build"

taskiq_aio_pika/broker.py

@@ -102,7 +102,7 @@ def __init__(
         self._label_for_routing = label_for_routing
         self._label_for_priority = label_for_priority
 
-        self._delay_queue = delay_queue or Queue(name="taskiq.delay")
+        self._delay_queue = delay_queue
 
         self._delayed_message_exchange_plugin = delayed_message_exchange_plugin
         if self._delayed_message_exchange_plugin:
@@ -261,15 +261,15 @@ async def _declare_queues(
             self._task_queues.append(Queue())
 
         queues = self._task_queues.copy()
-        if not self._delayed_message_exchange_plugin:
+        if not self._delayed_message_exchange_plugin and self._delay_queue:
             queues.append(self._delay_queue)
 
         for queue in filter(lambda queue: queue.declare, queues):
             per_queue_arguments: FieldTable = queue_default_arguments.copy()
             if queue.max_priority is not None:
                 per_queue_arguments["x-max-priority"] = queue.max_priority
             per_queue_arguments["x-queue-type"] = queue.type.value
-            if queue.name == self._delay_queue.name:
+            if self._delay_queue and queue.name == self._delay_queue.name:
                 per_queue_arguments["x-dead-letter-exchange"] = self._exchange.name
                 per_queue_arguments["x-dead-letter-routing-key"] = (
                     self._delay_queue.routing_key
@@ -292,7 +292,7 @@ async def _declare_queues(
                 "Bind queue to exchange with routing key '%s'",
                 queue.routing_key or queue.name,
             )
-            if queue.name != self._delay_queue.name:
+            if not self._delay_queue or queue.name != self._delay_queue.name:
                 await declared_queue.bind(
                     exchange=self._exchange.name,
                     routing_key=queue.routing_key or queue.name,
@@ -401,12 +401,17 @@ async def kick(self, message: BrokerMessage) -> None:
                 self._delayed_message_exchange.name,
             )
             await exchange.publish(rmq_message, routing_key=routing_key_name)
-        else:
+        elif self._delay_queue:
             rmq_message.expiration = timedelta(seconds=delay)
             await self.write_channel.default_exchange.publish(
                 rmq_message,
                 routing_key=self._delay_queue.routing_key or self._delay_queue.name,
             )
+        else:
+            raise IncorrectRoutingKeyError(
+                "Delay requested but no delay queue or delayed-message-exchange "
+                "is configured in the broker.",
+            )
 
     async def listen(self) -> AsyncGenerator[AckableMessage, None]:
         """
@@ -442,7 +447,7 @@ async def body(
             *[
                 body(queue, consumer_args)
                 for queue, consumer_args in queue_with_consumer_args_list
-                if queue.name != self._delay_queue.name
+                if not self._delay_queue or queue.name != self._delay_queue.name
             ],
         )
 

tests/test_routing.py

@@ -23,14 +23,15 @@ async def cleanup_class_broker(self, amqp_url: str) -> AsyncGenerator[None, None
         yield
         if self.broker is not None:
             await self.broker.shutdown()
+            queue_names = [queue.name for queue in self.broker._task_queues] + [
+                self.broker._dead_letter_queue.name,
+            ]
+            if self.broker._delay_queue is not None:
+                queue_names.append(self.broker._delay_queue.name)
             await _cleanup_amqp_resources(
                 amqp_url,
                 [self.broker._exchange.name],
-                [queue.name for queue in self.broker._task_queues]
-                + [
-                    self.broker._dead_letter_queue.name,
-                    self.broker._delay_queue.name,
-                ],
+                queue_names,
             )
 
     async def test_when_message_has_wrong_format__then_message_still_can_be_received(

tests/test_startup.py

@@ -20,14 +20,15 @@ async def cleanup_class_broker(self, amqp_url: str) -> AsyncGenerator[None, None
         yield
         if self.broker is not None:
             await self.broker.shutdown()
+            queue_names = [queue.name for queue in self.broker._task_queues] + [
+                self.broker._dead_letter_queue.name,
+            ]
+            if self.broker._delay_queue is not None:
+                queue_names.append(self.broker._delay_queue.name)
             await _cleanup_amqp_resources(
                 amqp_url,
                 [self.broker._exchange.name],
-                [queue.name for queue in self.broker._task_queues]
-                + [
-                    self.broker._dead_letter_queue.name,
-                    self.broker._delay_queue.name,
-                ],
+                queue_names,
             )
 
     async def test_when_declare_flag_passed_to_queue__broker_declare_queue_on_startup(
@@ -227,3 +228,26 @@ async def test_when_delayed_message_exchange_plugin_enabled_and_custom_exchange_
             match=f"Exchange '{delayed_message_exchange_name}' was not declared and does not exist.",
         ):
             await self.broker.startup()
+
+    async def test_when_delay_queue_not_specified__broker_does_not_create_delay_queue(
+        self,
+        amqp_url: str,
+        test_channel: Channel,
+        exchange_name: str,
+    ) -> None:
+        # given
+        self.broker = AioPikaBroker(
+            url=amqp_url,
+            exchange=Exchange(
+                name=exchange_name,
+                declare=True,
+            ),
+        )
+
+        # when
+        await self.broker.startup()
+
+        # then
+        assert self.broker._delay_queue is None
+        with pytest.raises(aiormq.exceptions.ChannelNotFoundEntity):
+            await test_channel.get_queue("taskiq.delay", ensure=True)