Implement Redis-backed WebSocket Pub/Sub helpers

docs/api/reference.md

@@ -39,10 +39,10 @@ async def handler(redis: AsyncRedisDep):
 ### `get_async_redis()`
 
 ```python
-async def get_async_redis(request: Request) -> AsyncRedis | AsyncRedisCluster
+async def get_async_redis(connection: HTTPConnection) -> AsyncRedis | AsyncRedisCluster
 ```
 
-Async function underlying `AsyncRedisDep`. Returns the same client instance on every call (no per-request overhead). Raises `RuntimeError` if no lifespan has initialised the pool.
+Async function underlying `AsyncRedisDep`. Accepts both HTTP ``Request`` and ``WebSocket`` (``HTTPConnection`` is their common base), so it works in both endpoint types. Returns the same client instance on every call (no per-request overhead). Raises ``RuntimeError`` if no lifespan has initialised the pool.
 
 ---
 
@@ -169,3 +169,52 @@ def default_key_builder(request: Request, eviction_group: str = "", prefix: str
 ```
 
 Builds a cache key from the request path (slashes → colons) and sorted query params. Eviction group is wrapped in Redis hash-tag braces (`{eviction_group}`) for Cluster slot consistency.
+
+---
+
+## Pub/Sub dependencies
+
+### `PubSubManagerDep`
+
+```python
+from redis_fastapi import PubSubManagerDep
+
+@app.post("/notify")
+async def notify(message: str, pubsub: PubSubManagerDep):
+    count = await pubsub.publish("notifications", message)
+    return {"subscribers": count}
+```
+
+`Annotated[PubSubManager, Depends(get_pubsub_manager)]` - publishes messages to Redis channels. Works in both HTTP endpoints and WebSocket handlers because ``get_async_redis`` accepts ``HTTPConnection``.
+
+### `get_pubsub_manager()`
+
+```python
+async def get_pubsub_manager(connection: HTTPConnection) -> PubSubManager
+```
+
+Factory function underlying `PubSubManagerDep`.
+
+### `PubSubManager`
+
+```python
+from redis_fastapi import PubSubManager
+
+pubsub = PubSubManager(redis)
+await pubsub.subscribe("channel")
+async for message in pubsub.listen():
+    ...
+await pubsub.close()
+```
+
+Wraps Redis Pub/Sub for real-time messaging. Uses a dedicated PubSub connection from the shared pool. For WebSocket endpoints, construct directly rather than using `PubSubManagerDep`.
+
+| Method | Description |
+|--------|-------------|
+| `subscribe(*channels)` | Subscribe to one or more Redis channels |
+| `unsubscribe(*channels)` | Unsubscribe from channels |
+| `publish(channel, message) -> int` | Publish a message, returns subscriber count |
+| `listen() -> AsyncIterator[dict]` | Yield `message`-type PubSub events |
+| `close()` | Unsubscribe all and close the connection |
+
+See the [Pub/Sub guide](../guide/pubsub.md) for WebSocket examples and best practices.

docs/guide/pubsub.md

@@ -0,0 +1,171 @@
+# Pub/Sub
+
+Redis [Pub/Sub](https://redis.io/docs/latest/develop/interact/pubsub/) enables
+real-time messaging between application instances.  fastapi-redis-sdk provides
+a `PubSubManager` that wraps Redis Pub/Sub for use in WebSocket endpoints and
+background tasks, reusing the shared async pool.
+
+## When to use Pub/Sub
+
+| Scenario | Example |
+|----------|---------|
+| Cross-worker chat rooms | Multiple server instances subscribe to the same channel |
+| Real-time notifications | Broadcast events to connected WebSocket clients |
+| Live updates | Push data changes to browsers without polling |
+| Broadcasts | Send the same message to all subscribers |
+
+## Setup
+
+`PubSubManager` only needs a Redis connection pool; no additional
+middleware or configuration is required:
+
+```python
+from fastapi import FastAPI
+from redis_fastapi import FastAPIRedis
+
+app = FastAPI()
+FastAPIRedis(app).lifespan()  # pool is set up; Pub/Sub is ready
+```
+
+## Publishing messages
+
+Use `PubSubManagerDep` from any HTTP endpoint to publish messages:
+
+```python
+from fastapi import Depends
+from redis_fastapi import PubSubManagerDep
+
+@app.post("/notify")
+async def notify(message: str, pubsub: PubSubManagerDep):
+    count = await pubsub.publish("notifications", message)
+    return {"subscribers": count}
+```
+
+The returned integer is the number of subscribers that received the message
+(across all connected Redis clients).
+
+## WebSocket subscriptions
+
+``PubSubManagerDep`` works in WebSocket handlers directly because
+``get_async_redis`` accepts ``HTTPConnection``, the common base type
+of both ``Request`` and ``WebSocket``:
+
+```python
+from fastapi import WebSocket
+from redis_fastapi import PubSubManagerDep
+
+@app.websocket("/ws/chat/{room}")
+async def chat(websocket: WebSocket, room: str, pubsub: PubSubManagerDep):
+    await websocket.accept()
+
+    channel = f"chat:{room}"
+    await pubsub.subscribe(channel)
+
+    async def forward_to_ws():
+        async for message in pubsub.listen():
+            await websocket.send_text(message["data"].decode())
+
+    async def forward_to_redis():
+        async for data in websocket.iter_text():
+            await pubsub.publish(channel, data)
+
+    try:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(forward_to_ws)
+            tg.start_soon(forward_to_redis)
+    except Exception:
+        pass
+    finally:
+        await pubsub.close()
+```
+
+The `listen()` method yields only `message`-type events; system messages
+(subscription confirmations, etc.) are filtered out automatically.
+
+## Broadcasting
+
+Publish to a channel without subscribing:
+
+```python
+@app.post("/broadcast")
+async def broadcast(message: str, pubsub: PubSubManagerDep):
+    count = await pubsub.publish("broadcasts", message)
+    return {"sent_to": count}
+```
+
+## Managing subscriptions
+
+Subscribe to multiple channels at once:
+
+```python
+await pubsub.subscribe("channel:a", "channel:b", "channel:c")
+```
+
+Unsubscribe selectively:
+
+```python
+await pubsub.unsubscribe("channel:a")
+```
+
+Unsubscribe all and close:
+
+```python
+await pubsub.close()
+```
+
+`close()` is safe to call multiple times and handles cleanup gracefully
+even if the underlying connection was never established.
+
+## Testing
+
+Unit tests can use `fakeredis` to simulate Pub/Sub without a real Redis
+server:
+
+```python
+import fakeredis.aioredis
+from redis_fastapi import PubSubManager
+
+async def test_pubsub():
+    fake = fakeredis.aioredis.FakeRedis()
+    pubsub = PubSubManager(fake)
+
+    # Subscribe and publish in sequence
+    await pubsub.subscribe("test:ch")
+
+    # Listen for the published message
+    async def publish_later():
+        import anyio
+        await anyio.sleep(0.05)
+        await pubsub.publish("test:ch", "hello")
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(publish_later)
+        async for message in pubsub.listen():
+            assert message["data"] == b"hello"
+            break
+
+    await pubsub.close()
+```
+
+## Limitations
+
+Redis Pub/Sub is **transient and at-most-once**:
+
+- Messages are **not persisted**; if a subscriber disconnects, it misses
+  messages sent while offline.
+- There is **no message acknowledgement**; if a subscriber crashes after
+  receiving a message, that message is lost.
+- Delivery is **at-most-once**; a message is delivered zero or one times.
+
+For persistent message queues with consumer groups, see
+[Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/).
+
+## API Reference
+
+| Method | Description |
+|--------|-------------|
+| `subscribe(*channels)` | Subscribe to one or more Redis channels |
+| `unsubscribe(*channels)` | Unsubscribe from Redis channels |
+| `publish(channel, message)` | Publish a message, returns subscriber count |
+| `listen()` | Async generator yielding `message`-type PubSub events |
+| `close()` | Unsubscribe all + close the PubSub connection |

examples/main.py

@@ -21,12 +21,14 @@
 
 from typing import Annotated
 
-from fastapi import Depends, FastAPI
+import anyio
+from fastapi import Depends, FastAPI, WebSocket
 
 from redis_fastapi import (
     AsyncRedisDep,
     CacheBackendDep,
     FastAPIRedis,
+    PubSubManagerDep,
     cache,
     cache_evict,
     default_key_builder,
@@ -127,3 +129,38 @@ async def delete_item(item_id: int, cache: CacheBackendDep) -> dict[str, object]
     """Evict a single item from the cache."""
     deleted = await cache.delete(f"item:{item_id}", eviction_group="items")
     return {"id": item_id, "deleted": deleted}
+
+
+@app.post("/notify")
+async def notify(message: str, pubsub: PubSubManagerDep) -> dict[str, int]:
+    """Publish a message to the ``notifications`` channel."""
+    count = await pubsub.publish("notifications", message)
+    return {"subscribers": count}
+
+
+@app.websocket("/ws/chat/{room}")
+async def chat(websocket: WebSocket, room: str, pubsub: PubSubManagerDep) -> None:
+    """WebSocket chat room backed by Redis Pub/Sub.
+
+    Connect: ``ws://localhost:8000/ws/chat/lobby``
+    """
+    await websocket.accept()
+    channel = f"chat:{room}"
+    await pubsub.subscribe(channel)
+
+    async def _send() -> None:
+        async for message in pubsub.listen():
+            await websocket.send_text(message["data"].decode())
+
+    async def _receive() -> None:
+        async for data in websocket.iter_text():
+            await pubsub.publish(channel, data)
+
+    try:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(_send)
+            tg.start_soon(_receive)
+    except Exception:
+        pass
+    finally:
+        await pubsub.close()

mkdocs.yml

@@ -82,6 +82,7 @@ nav:
   - User Guide:
     - Architecture: guide/architecture.md
     - Caching: guide/caching.md
+    - Pub/Sub: guide/pubsub.md
     - Configuration: guide/configuration.md
     - Benchmarks: guide/benchmarks.md
   - API Reference:

src/redis_fastapi/__init__.py

@@ -15,12 +15,15 @@
 from redis_fastapi.deps import (
     AsyncRedisDep,
     CacheBackendDep,
+    PubSubManagerDep,
     SyncCacheBackendDep,
     get_async_redis,
     get_cache_backend,
+    get_pubsub_manager,
     get_sync_cache_backend,
 )
 from redis_fastapi.lifespan import redis_lifespan
+from redis_fastapi.pubsub import PubSubManager
 from redis_fastapi.setup import FastAPIRedis
 from redis_fastapi.telemetry import disable_telemetry, enable_telemetry
 from redis_fastapi.types import (
@@ -39,6 +42,8 @@
     "JsonCoder",
     "KeyBuilder",
     "FastAPIRedis",
+    "PubSubManager",
+    "PubSubManagerDep",
     "RedisSettings",
     "SyncCacheBackend",
     "SyncCacheBackendDep",
@@ -51,6 +56,7 @@
     "enable_telemetry",
     "get_async_redis",
     "get_cache_backend",
+    "get_pubsub_manager",
     "get_settings",
     "get_sync_cache_backend",
     "pydantic_model_coder",

src/redis_fastapi/deps.py

@@ -7,11 +7,13 @@
 
 if TYPE_CHECKING:
     from redis_fastapi.cache_backend import CacheBackend, SyncCacheBackend
+    from redis_fastapi.pubsub import PubSubManager
 
-from fastapi import Depends, FastAPI, Request
+from fastapi import Depends, FastAPI
 from redis.asyncio import ConnectionPool as AsyncConnectionPool
 from redis.asyncio import Redis as AsyncRedis
 from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
+from starlette.requests import HTTPConnection, Request
 
 from redis_fastapi.config import get_settings
 
@@ -103,9 +105,13 @@ def _get_pool_state(app: FastAPI) -> _PoolState:
     return state
 
 
-async def get_async_redis(request: Request) -> AsyncClient:
+async def get_async_redis(connection: HTTPConnection) -> AsyncClient:
     """Return an async Redis client backed by the shared connection pool.
 
+    Accepts both HTTP ``Request`` and ``WebSocket`` instances (both
+    inherit from ``HTTPConnection``), so this dependency works in
+    HTTP endpoints and WebSocket handlers alike.
+
     Returns a cached client instance - the same wrapper is reused
     across calls to avoid per-request overhead.
 
@@ -114,7 +120,7 @@ async def get_async_redis(request: Request) -> AsyncClient:
     Raises:
         RuntimeError: If no lifespan has initialized the pool.
     """
-    return _get_pool_state(request.app).get_async_client()
+    return _get_pool_state(connection.app).get_async_client()
 
 
 async def get_cache_backend(request: Request) -> CacheBackend:
@@ -138,6 +144,21 @@ async def get_sync_cache_backend(request: Request) -> SyncCacheBackend:
     return SyncCacheBackend(backend)
 
 
+async def get_pubsub_manager(
+    redis: AsyncClient = Depends(get_async_redis),
+) -> PubSubManager:
+    """Return a :class:`PubSubManager` backed by the shared async pool.
+
+    Works in both HTTP endpoints and WebSocket handlers because
+    ``get_async_redis`` accepts ``HTTPConnection``, the common base
+    of ``Request`` and ``WebSocket``.
+    """
+    from redis_fastapi.pubsub import PubSubManager  # noqa: WPS433
+
+    return PubSubManager(redis)
+
+
 AsyncRedisDep = Annotated[AsyncClient, Depends(get_async_redis)]
 CacheBackendDep = Annotated["CacheBackend", Depends(get_cache_backend)]
 SyncCacheBackendDep = Annotated["SyncCacheBackend", Depends(get_sync_cache_backend)]
+PubSubManagerDep = Annotated["PubSubManager", Depends(get_pubsub_manager)]