chore: typing

Author: KatantDev

taskiq/receiver/receiver.py

@@ -4,7 +4,7 @@
 import inspect
 import random
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from concurrent.futures import Executor, ProcessPoolExecutor
 from logging import getLogger
 from time import time
@@ -234,7 +234,7 @@ async def callback(
     async def batched_callback(  # noqa: C901
         self,
         task_name: str,
-        messages: "list[bytes | AckableMessage]",
+        messages: "Sequence[bytes | AckableMessage]",
     ) -> None:
         """
         Execute a batch of accumulated messages as a single call.
@@ -637,7 +637,7 @@ def _peek_task_name(self, message: "bytes | AckableMessage") -> str | None:
     async def _flush_batch(
         self,
         task_name: str,
-        messages: "list[bytes | AckableMessage]",
+        messages: "Sequence[bytes | AckableMessage]",
     ) -> None:
         """
         Flush handler invoked by the Batcher: run a batch as one task.
@@ -689,9 +689,7 @@ async def _dispatch_message(
         :param task_cb: done-callback that releases the semaphore for a task.
         """
         batched_name = self._peek_task_name(message)
-        cfg = (
-            self.get_batch_config(batched_name) if batched_name is not None else None
-        )
+        cfg = self.get_batch_config(batched_name) if batched_name is not None else None
         if batched_name is not None and cfg is not None:
             size, timeout = cfg
             # Buffering is not execution: release the slot acquired for this

tests/brokers/test_inmemory.py

@@ -180,4 +180,3 @@ async def batched(items: list[int]) -> int:
     results = [await t.wait_result(timeout=2) for t in tasks]
 
     assert all(r.is_err for r in results)
-

tests/receiver/test_batch_callback.py

@@ -82,6 +82,7 @@ async def my_task(items: list[int]) -> int:
     def make_ack(i: int) -> Callable[[], None]:
         def _ack() -> None:
             acked.append(i)
+
         return _ack
 
     msgs = [
@@ -127,7 +128,7 @@ async def my_task(items: list[int]) -> int:
 
     receiver = _receiver(broker)
 
-    queue: asyncio.Queue = asyncio.Queue()
+    queue: asyncio.Queue[bytes | AckableMessage] = asyncio.Queue()
     for i in (1, 2):
         await queue.put(_msg(broker, my_task.task_name, i))
     await queue.put(QUEUE_DONE)
@@ -149,7 +150,7 @@ async def my_task(items: list[int]) -> int:
 
     receiver = _receiver(broker)
 
-    queue: asyncio.Queue = asyncio.Queue()
+    queue: asyncio.Queue[bytes | AckableMessage] = asyncio.Queue()
     for i in (1, 2):
         await queue.put(_msg(broker, my_task.task_name, i))
     await queue.put(QUEUE_DONE)
@@ -171,7 +172,7 @@ async def normal_task(x: int) -> int:
 
     receiver = _receiver(broker)
 
-    queue: asyncio.Queue = asyncio.Queue()
+    queue: asyncio.Queue[bytes | AckableMessage] = asyncio.Queue()
     await queue.put(_msg(broker, normal_task.task_name, 7))
     await queue.put(QUEUE_DONE)
 
@@ -246,6 +247,7 @@ async def my_task(items: list[int]) -> int:
     def make_ack(i: int) -> Callable[[], None]:
         def _ack() -> None:
             acked.append(i)
+
         return _ack
 
     msgs = [
@@ -274,6 +276,7 @@ async def my_task(items: list[int]) -> int:
     def make_ack(i: int) -> Callable[[], None]:
         def _ack() -> None:
             acked.append(i)
+
         return _ack
 
     msgs = [
@@ -309,7 +312,7 @@ async def my_task(items: list[int]) -> int:
     # Only one concurrent slot.
     receiver = _receiver(broker, max_async_tasks=1)
 
-    queue: asyncio.Queue = asyncio.Queue()
+    queue: asyncio.Queue[bytes | AckableMessage] = asyncio.Queue()
     await queue.put(_msg(broker, blocker.task_name, 99))
     # These 3 batched messages must buffer and flush even though the single
     # semaphore slot is held by `blocker`.