redis/redis-py · error · ValueError

PubSub is not supported for RedisCluster

Error message

PubSub is not supported for RedisCluster

What it means

Raised as ValueError by DefaultCommandExecutor.pubsub() (command_executor.py:222-224) when pub/sub is first requested and the active database's underlying client is a RedisCluster. The multi-database PubSub wrapper delegates to a single active connection and is not built to fan out cluster pub/sub subscriptions, so it refuses creation rather than producing partial/inconsistent subscriptions.

Solutions

1. Use standalone redis.asyncio.Redis databases (client_class=Redis) in the MultiDBClient if pub/sub is required.
2. For cluster pub/sub, operate a separate dedicated RedisCluster client outside MultiDBClient.
3. Before subscribing, verify the active database is non-cluster: assert not isinstance(client.command_executor.active_database.client, RedisCluster).
4. Design topology so the pub/sub-capable database has the highest weight and is active at startup.

Example fix

// before
config = MultiDbConfig(
    databases_config=[DatabaseConfig(from_url="redis://cluster:6379", client_kwargs={...})],
)
client = MultiDbConfig(config); client._config.client_class = RedisCluster
pubsub = await client.pubsub()  # ValueError

// after - use a standalone Redis for the pub/sub-capable database
config = MultiDbConfig(
    databases_config=[DatabaseConfig(from_url="redis://standalone:6379")],
    client_class=Redis,
)
client = MultiDBClient(config)
pubsub = await client.pubsub()

from redis.asyncio import RedisCluster

def active_supports_pubsub(client) -> bool:
    active = client.command_executor.active_database
    return active is not None and not isinstance(active.client, RedisCluster)

# before subscribing:
if not active_supports_pubsub(client):
    raise NotImplementedError("pub/sub requires a non-cluster active database")
pubsub = await client.pubsub()

from redis.asyncio import Redis, RedisCluster

def active_is_standalone(client) -> bool:
    active = client.command_executor.active_database
    return active is not None and isinstance(active.client, Redis)

try:
    pubsub = await client.pubsub()
except ValueError as e:
    if "PubSub is not supported for RedisCluster" in str(e):
        # route pub/sub to a dedicated standalone Redis client instead
        pubsub = standalone_redis.pubsub()
    else:
        raise

When it happens

Trigger: Calling await client.pubsub(...) on a MultiDBClient whose active database was configured with client_class=RedisCluster (or a cluster URL), or after failover promoted a cluster database to active. The check fires the first time pubsub() is called (when self._active_pubsub is None).

Common situations: Mixing Redis Cluster and standalone Redis in an Active-Active topology and trying to use pub/sub; pointing a DatabaseConfig at a redis://cluster URL with RedisCluster as client_class and then subscribing; failover landing on a cluster member.

Related errors

• Cluster client has no nodes - cannot create health check…
• HIMPORT is not supported on the multi-database…
• Node : doesn't exist in the cluster
• Passing a host requires passing a port, and vice versa
• Cannot set active database, database is unhealthy

AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10). Data as JSON: /api/errors/06f569361e5739d4. Report an issue: GitHub.

Appendix: source

Thrown at redis/asyncio/multidb/command_executor.py:224

    def active_pubsub(self) -> Optional[PubSub]:
        return self._active_pubsub

    @active_pubsub.setter
    def active_pubsub(self, pubsub: PubSub) -> None:
        self._active_pubsub = pubsub

    @property
    def failover_strategy_executor(self) -> FailoverStrategyExecutor:
        return self._failover_strategy_executor

    @property
    def command_retry(self) -> Retry:
        return self._command_retry

    def pubsub(self, **kwargs):
        if self._active_pubsub is None:
            if isinstance(self._active_database.client, RedisCluster):
                raise ValueError("PubSub is not supported for RedisCluster")

            self._active_pubsub = self._active_database.client.pubsub(**kwargs)
            self._active_pubsub_kwargs = kwargs

    async def execute_command(self, *args, **options):
        async def callback():
            response = await self._active_database.client.execute_command(
                *args, **options
            )
            await self._register_command_execution(args)
            return response

        return await self._execute_with_failure_detection(callback, args)

    async def execute_pipeline(self, command_stack: tuple):
        async def callback():
            async with self._active_database.client.pipeline() as pipe:
                for command, options in command_stack:

View on GitHub (pinned to 6a6b581b48)