Feat/redis pipeline (#2270)

* example tests fail

* lint

* fix: change incr to len(result)

* feat(broker): add 'pipeline' parameter to publish function

* fix(tests): piblisher.publish & Event.set

* docs: add example redis pipeline

* fix: raise exception on both param rpc and pipeline

* docs: add redis/pipeline.md

* fix: update hint abount type of queue

* fix(docs): removed snippets

* docs: moved from examples to docs_src

* docs: add info about publish_batch

* docs: autogenerated

* fix: raise in-place exception

* feat: add pipeline for publish_batch

* fix:  get_pipline renamed to get_pipe

* add tests for docs_src and publish_batch

* docs: generate API References

* fix: ruff lint

* fix: removed mark.pipe

* tests: polish tests examples

* tests: 3.8 compatible

* docs: polish pipeline docs

* docs: generate API References

---------

Co-authored-by: Nikita Pastukhov <[email protected]>
Co-authored-by: Lancetnik <[email protected]>

Author: suiseriff

docs/docs/SUMMARY.md

@@ -119,6 +119,7 @@ search:
         - [Batching](redis/streams/batch.md)
         - [Acknowledgement](redis/streams/ack.md)
     - [RPC](redis/rpc.md)
+    - [Pipeline](redis/pipeline.md)
     - [Message Information](redis/message.md)
     - [Security Configuration](redis/security.md)
 - [Reference - Code API](api/index.md)
@@ -1021,6 +1022,8 @@ search:
                 - [StreamSub](api/faststream/redis/StreamSub.md)
                 - [TestApp](api/faststream/redis/TestApp.md)
                 - [TestRedisBroker](api/faststream/redis/TestRedisBroker.md)
+                - annotations
+                    - [get_pipe](api/faststream/redis/annotations/get_pipe.md)
                 - broker
                     - broker
                         - [RedisBroker](api/faststream/redis/broker/broker/RedisBroker.md)

docs/docs/en/api/faststream/redis/annotations/get_pipe.md

@@ -0,0 +1,11 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 0.5
+---
+
+::: faststream.redis.annotations.get_pipe

docs/docs/en/redis/pipeline.md

@@ -0,0 +1,47 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 10
+---
+
+# Redis Pipeline
+
+**FastStream** supports [**Redis** pipelining](https://redis.io/docs/latest/develop/use/pipelining/){.external-link target="_blank"} to optimize performance when publishing multiple messages in a batch. This allows you to queue several **Redis** operations and execute them in one network round-trip, reducing latency significantly.
+
+## Usage Example
+
+```python linenums="1" hl_lines="2 11 19 22"
+{! docs_src/redis/pipeline/pipeline.py !}
+```
+
+## API
+
+You can pass the `pipeline` parameter to the `publish` method to delay the execution of **Redis** commands. The commands will only be executed after you explicitly call `#!python await pipe.execute()`.
+
+The `pipeline` object is injected by the `Pipeline` annotation:
+
+```python
+from faststream.redis.annotations import Pipeline
+```
+
+`Pipeline` is a **Redis** pipeline object (`redis.asyncio.client.Pipeline`), which is wrapped in a FastStream dependency and will be automatically available in any subscriber.
+
+## Batch Publishing with Pipeline
+
+When using `#!python broker.publish_batch()` in combination with the `pipeline` parameter, all messages sent through the pipeline are queued and processed by the subscriber as a single batch after calling `#!python await pipe.execute()`. This allows the subscriber to handle all messages sent through the pipeline in a single execution, improving the efficiency of batch processing.
+
+## Notes
+
+- Pipelining is supported for all **Redis** queue types, including channels, lists, and streams.
+- You can combine multiple queue types in a single pipeline.
+
+## Benefits
+
+- Reduces network traffic by batching **Redis** commands.
+- Improves performance in high-volume scenarios.
+- Fully integrates with **FastStream**'s dependency injection system.
+- Allows for efficient batch processing when using `#!python broker.publish_batch()` and `pipeline`, as all messages are processed as a single entity by the subscriber after `#!python await pipe.execute()`.

docs/docs/navigation_template.txt

@@ -119,6 +119,7 @@ search:
         - [Batching](redis/streams/batch.md)
         - [Acknowledgement](redis/streams/ack.md)
     - [RPC](redis/rpc.md)
+    - [Pipeline](redis/pipeline.md)
     - [Message Information](redis/message.md)
     - [Security Configuration](redis/security.md)
 - [Reference - Code API](api/index.md)

docs/docs_src/redis/pipeline/__init__.py

@@ -0,0 +1,27 @@
+from faststream import FastStream, Logger
+from faststream.redis import RedisBroker, Pipeline
+
+broker = RedisBroker()
+app = FastStream(broker)
+
+@broker.subscriber("test")
+async def handle(
+    msg: str,
+    logger: Logger,
+    pipe: Pipeline,
+) -> None:
+    logger.info(msg)
+
+    for i in range(10):
+        await broker.publish(
+            f"hello {i}",
+            channel="test-output",  # queue can be channel, list, or stream
+            pipeline=pipe,
+        )
+
+    results = await pipe.execute()  # execute all publish commands
+    logger.info(results)
+
+@app.after_startup
+async def t() -> None:
+    await broker.publish("Hi!", "test")

docs/docs_src/redis/pipeline/pipeline.py

@@ -1,7 +1,7 @@
 try:
     from faststream.testing.app import TestApp
 
-    from .annotations import Redis, RedisMessage
+    from .annotations import Pipeline, Redis, RedisMessage
     from .broker.broker import RedisBroker
     from .response import RedisResponse
     from .router import RedisPublisher, RedisRoute, RedisRouter
@@ -15,6 +15,7 @@
 
 __all__ = (
     "ListSub",
+    "Pipeline",
     "PubSub",
     "Redis",
     "RedisBroker",

faststream/redis/__init__.py

@@ -1,15 +1,28 @@
-from redis.asyncio.client import Redis as RedisClient
+from typing import TYPE_CHECKING, AsyncIterator
+
+from redis.asyncio.client import Pipeline as _RedisPipeline
+from redis.asyncio.client import Redis as _RedisClient
 from typing_extensions import Annotated
 
+from faststream import Depends
 from faststream.annotations import ContextRepo, Logger, NoCast
 from faststream.redis.broker.broker import RedisBroker as RB
 from faststream.redis.message import UnifyRedisMessage
 from faststream.utils.context import Context
 
+if TYPE_CHECKING:
+    RedisClient = _RedisClient[bytes]
+    RedisPipeline = _RedisPipeline[bytes]
+else:
+    RedisClient = _RedisClient
+    RedisPipeline = _RedisPipeline
+
+
 __all__ = (
     "ContextRepo",
     "Logger",
     "NoCast",
+    "Pipeline",
     "Redis",
     "RedisBroker",
     "RedisMessage",
@@ -18,3 +31,11 @@
 RedisMessage = Annotated[UnifyRedisMessage, Context("message")]
 RedisBroker = Annotated[RB, Context("broker")]
 Redis = Annotated[RedisClient, Context("broker._connection")]
+
+
+async def get_pipe(redis: Redis) -> AsyncIterator[RedisPipeline]:
+    async with redis.pipeline() as pipe:
+        yield pipe
+
+
+Pipeline = Annotated[RedisPipeline, Depends(get_pipe, cast=False)]

faststream/redis/annotations.py

@@ -16,7 +16,7 @@
 
 import anyio
 from anyio import move_on_after
-from redis.asyncio.client import Redis
+from redis.asyncio.client import Pipeline, Redis
 from redis.asyncio.connection import (
     Connection,
     ConnectionPool,
@@ -451,6 +451,13 @@ async def publish(  # type: ignore[override]
                 "Argument will be removed in **FastStream 0.6.0**."
             ),
         ] = False,
+        pipeline: Annotated[
+            Optional["Pipeline[bytes]"],
+            Doc(
+                "Optional Redis `Pipeline` object to batch multiple commands. "
+                "Use it to group Redis operations for optimized execution and reduced latency."
+            ),
+        ] = None,
     ) -> Optional["DecodedMessage"]:
         """Publish message directly.
 
@@ -472,6 +479,7 @@ async def publish(  # type: ignore[override]
             rpc=rpc,
             rpc_timeout=rpc_timeout,
             raise_timeout=raise_timeout,
+            pipeline=pipeline,
         )
 
     @override
@@ -517,6 +525,13 @@ async def publish_batch(
                 "**correlation_id** is a useful option to trace messages."
             ),
         ] = None,
+        pipeline: Annotated[
+            Optional["Pipeline[bytes]"],
+            Doc(
+                "Optional Redis `Pipeline` object to batch multiple commands. "
+                "Use it to group Redis operations for optimized execution and reduced latency."
+            ),
+        ] = None,
     ) -> None:
         """Publish multiple messages to Redis List by one request."""
         assert self._producer, NOT_CONNECTED_YET  # nosec B101
@@ -532,6 +547,7 @@ async def publish_batch(
             *msgs,
             list=list,
             correlation_id=correlation_id,
+            pipeline=pipeline,
         )
 
     @override

faststream/redis/broker/broker.py

@@ -13,7 +13,7 @@
 from faststream.utils.nuid import NUID
 
 if TYPE_CHECKING:
-    from redis.asyncio.client import PubSub, Redis
+    from redis.asyncio.client import Pipeline, PubSub, Redis
 
     from faststream.broker.types import (
         AsyncCallable,
@@ -62,10 +62,17 @@ async def publish(  # type: ignore[override]
         rpc: bool = False,
         rpc_timeout: Optional[float] = 30.0,
         raise_timeout: bool = False,
+        pipeline: Optional["Pipeline[bytes]"] = None,
     ) -> Optional[Any]:
         if not any((channel, list, stream)):
             raise SetupError(INCORRECT_SETUP_MSG)
 
+        if pipeline is not None and rpc is True:
+            raise RuntimeError(
+                "You cannot use both rpc and pipeline arguments at the same time: "
+                "select only one delivery mechanism."
+            )
+
         psub: Optional[PubSub] = None
         if rpc:
             if reply_to:
@@ -83,12 +90,13 @@ async def publish(  # type: ignore[override]
             correlation_id=correlation_id,
         )
 
+        conn = pipeline or self._connection
         if channel is not None:
-            await self._connection.publish(channel, msg)
+            await conn.publish(channel, msg)
         elif list is not None:
-            await self._connection.rpush(list, msg)
+            await conn.rpush(list, msg)
         elif stream is not None:
-            await self._connection.xadd(
+            await conn.xadd(
                 name=stream,
                 fields={DATA_KEY: msg},
                 maxlen=maxlen,
@@ -193,6 +201,7 @@ async def publish_batch(
         list: str,
         correlation_id: str,
         headers: Optional["AnyDict"] = None,
+        pipeline: Optional["Pipeline[bytes]"] = None,
     ) -> None:
         batch = (
             RawMessage.encode(
@@ -203,4 +212,5 @@ async def publish_batch(
             )
             for msg in msgs
         )
-        await self._connection.rpush(list, *batch)
+        conn = pipeline or self._connection
+        await conn.rpush(list, *batch)