From f4d08c459d6a5a958d87f24a3d8c63d3fae7ef11 Mon Sep 17 00:00:00 2001 From: Olga Pichuzhkina Date: Sat, 26 Sep 2026 04:57:22 +0300 Subject: [PATCH 1/2] fix(redis): stop ListBatchPublisher from passing as a ListPublisher ListBatchPublisher subclassed ListPublisher while replacing `publish(message, list, ...)` with `publish(*messages, list=...)`. Code typed against ListPublisher, including the `ListPublisher | ListBatchPublisher` union that `publisher(list=ListSub(q, batch=flag))` returns, type checked `pub.publish("hello", "other")`. On a batch publisher that call sends ["hello", "other"] as one batch to the publisher's own list. The shared part (constructor, `list`, `subscriber_property`, `request`) moves to `_ListPublisherMixin`, and both publishers derive from it, the way `_ListHandlerMixin` already splits the list subscribers. No method changes; `isinstance(batch_publisher, ListPublisher)` is now False. pyrefly reported it as bad-override on `ListBatchPublisher.publish`, the one override among its findings that a user can hit (#3202). --- faststream/redis/publisher/usecase.py | 62 +++++++++++++++------------ tests/mypy/redis.py | 9 ++++ 2 files changed, 43 insertions(+), 28 deletions(-) diff --git a/faststream/redis/publisher/usecase.py b/faststream/redis/publisher/usecase.py index dd4bc504c3..206f3a091e 100644 --- a/faststream/redis/publisher/usecase.py +++ b/faststream/redis/publisher/usecase.py @@ -166,7 +166,7 @@ async def request( return msg -class ListPublisher(LogicPublisher): +class _ListPublisherMixin(LogicPublisher): __slots__ = ("_list",) def __init__( @@ -192,6 +192,36 @@ def subscriber_property(self, *, name_only: bool) -> dict[str, Any]: "stream": None, } + @override + async def request( + self, + message: "SendableMessage" = None, + list: str | None = None, + *, + correlation_id: str | None = None, + headers: dict[str, Any] | None = None, + timeout: float | None = 30.0, + ) -> "RedisChannelMessage": + cmd = RedisPublishCommand( + message, + list=list or self.list.name, + headers=self.headers | (headers or {}), + correlation_id=correlation_id or self._outer_config.id_generator(), + timeout=timeout, + _publish_type=PublishType.REQUEST, + message_format=self.config.message_format, + ) + + msg: RedisChannelMessage = await self._basic_request( + cmd, + producer=self.producer, + ) + return msg + + +class ListPublisher(_ListPublisherMixin): + __slots__ = () + @override async def publish( self, @@ -242,34 +272,10 @@ async def _publish( _extra_middlewares=_extra_middlewares, ) - @override - async def request( - self, - message: "SendableMessage" = None, - list: str | None = None, - *, - correlation_id: str | None = None, - headers: dict[str, Any] | None = None, - timeout: float | None = 30.0, - ) -> "RedisChannelMessage": - cmd = RedisPublishCommand( - message, - list=list or self.list.name, - headers=self.headers | (headers or {}), - correlation_id=correlation_id or self._outer_config.id_generator(), - timeout=timeout, - _publish_type=PublishType.REQUEST, - message_format=self.config.message_format, - ) - - msg: RedisChannelMessage = await self._basic_request( - cmd, - producer=self.producer, - ) - return msg - -class ListBatchPublisher(ListPublisher): +# Not a `ListPublisher`: its `publish(*messages)` would take the `list` that +# `ListPublisher.publish(message, list)` accepts positionally as a second message. +class ListBatchPublisher(_ListPublisherMixin): __slots__ = () @override diff --git a/tests/mypy/redis.py b/tests/mypy/redis.py index 439d70ff1d..1e72516f2b 100644 --- a/tests/mypy/redis.py +++ b/tests/mypy/redis.py @@ -536,3 +536,12 @@ def accepts_any_list_sub(list_sub: ListSub) -> None: ... accepts_any_list_sub(ListSub("test", batch=True)) + + +def accepts_list_publisher(publisher: ListPublisher) -> None: ... + + +# `publish(*messages)` of a batch publisher would take the `list` passed +# positionally to `ListPublisher.publish` as a second message +batch_publisher = RedisBroker().publisher(list=ListSub("test", batch=True)) +accepts_list_publisher(batch_publisher) # type: ignore[arg-type] From acbd36b09727e7523164b69f02dd2d7aeeb1c966 Mon Sep 17 00:00:00 2001 From: Olga Pichuzhkina Date: Sat, 26 Sep 2026 05:28:18 +0300 Subject: [PATCH 2/2] test(redis): cover request() on a list publisher Moving `request()` into `_ListPublisherMixin` counts its lines as changed, and diff-cover found no test that ever calls it: the shared request testcase only builds a channel publisher for redis. The new case runs in memory and against a real Redis. --- tests/brokers/redis/test_requests.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/tests/brokers/redis/test_requests.py b/tests/brokers/redis/test_requests.py index 0d56c8b106..bc56f0fe9d 100644 --- a/tests/brokers/redis/test_requests.py +++ b/tests/brokers/redis/test_requests.py @@ -4,7 +4,7 @@ import pytest from faststream import BaseMiddleware -from faststream.redis import BinaryMessageFormatV1 +from faststream.redis import BinaryMessageFormatV1, RedisBroker from tests.brokers.base.requests import RequestsTestcase from .basic import RedisMemoryTestcaseConfig, RedisTestcaseConfig @@ -34,6 +34,22 @@ class RedisRequestsTestcase(RequestsTestcase): def get_middleware(self, **kwargs: Any) -> Any: return Mid + async def test_list_publisher_request(self, queue: str) -> None: + broker: RedisBroker = self.get_broker() + + publisher = broker.publisher(list=queue) + + @broker.subscriber(list=queue) + async def handler(msg: Any) -> str: + return "Response" + + async with self.patch_broker(broker): + await broker.start() + + response = await publisher.request(None, timeout=self.timeout) + + assert await response.decode() == "Response" + @pytest.mark.connected() @pytest.mark.redis()