redis/redis-py · error · TypeError

Unsupported client type

Error message

Unsupported client type: {type(database.client)}

What it means

Raised as TypeError by AbstractHealthCheckPolicy.get_client() (healthcheck.py:212-213) when the database's client is none of the four supported types (AsyncRedis, SyncRedis, AsyncRedisCluster, SyncRedisCluster). The health-check layer only knows how to derive connection kwargs / startup nodes for those, so an unrecognised client type is rejected rather than probed incorrectly.

Solutions

  1. Use one of the supported client classes (redis.asyncio.Redis or redis.asyncio.RedisCluster) as the database client.
  2. If you need a subclass, inherit from AsyncRedis/AsyncRedisCluster so isinstance checks pass.
  3. In tests, use a real AsyncRedis against a test container or fakeredis that subclasses AsyncRedis, not an unrelated Mock.
  4. Inspect the message's type(...) to find which database holds the unsupported client and fix its DatabaseConfig.

Example fix

// before
class MyRedis:
    # does not subclass AsyncRedis
    ...
db_cfg = DatabaseConfig(client_kwargs={})  # MultiDbConfig.client_class = MyRedis
# health check raises TypeError

// after
from redis.asyncio import Redis
class MyRedis(Redis):
    ...
db_cfg = DatabaseConfig(client_kwargs={"host": "redis.local", "port": 6379})
config = MultiDbConfig(databases_config=[db_cfg], client_class=MyRedis)
Defensive patterns

Strategy: type-guard

Validate before calling

from redis.asyncio import Redis as AsyncRedis, RedisCluster as AsyncRedisCluster
from redis.client import Redis as SyncRedis
from redis.cluster import RedisCluster as SyncRedisCluster

def client_type_supported(db) -> bool:
    return isinstance(db.client, (AsyncRedis, SyncRedis, AsyncRedisCluster, SyncRedisCluster))

# before initialize():
for db, _ in client.get_databases():
    assert client_type_supported(db), f"unsupported client type {type(db.client)!r}"

Type guard

from redis.asyncio import Redis as AsyncRedis, RedisCluster as AsyncRedisCluster
from redis.client import Redis as SyncRedis
from redis.cluster import RedisCluster as SyncRedisCluster

def is_supported_client(client) -> bool:
    return isinstance(client, (AsyncRedis, SyncRedis, AsyncRedisCluster, SyncRedisCluster))

Try / catch

try:
    await client.initialize()
except TypeError as e:
    if "Unsupported client type" in str(e):
        # swap the offending DatabaseConfig to use redis.asyncio.Redis / RedisCluster
        raise RuntimeError("use redis.asyncio.Redis or RedisCluster as the database client")
    raise

Prevention

When it happens

Trigger: Setting MultiDbConfig.client_class (or a Database's client) to a custom/non-standard client class — a third-party Redis wrapper, a mock/stub in tests, or a subclass the policy cannot introspect. The f-string includes type(database.client) to identify the offender.

Common situations: Injecting a test double/mock client that is not a Redis subclass; using a custom Redis subclass that does not inherit from one of the supported bases; wrapping the client in a proxy object; version skew where a refactor changed the client class hierarchy.

Related errors


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

Appendix: source

Thrown at redis/asyncio/multidb/healthcheck.py:213

                        nodes_manager,
                        "require_full_coverage",
                        getattr(nodes_manager, "_require_full_coverage", True),
                    )
                    client = AsyncRedisCluster(
                        host=first_node.host,
                        port=first_node.port,
                        dynamic_startup_nodes=nodes_manager._dynamic_startup_nodes,
                        address_remap=nodes_manager.address_remap,
                        require_full_coverage=require_full_coverage,
                        retry=database.client.retry,
                        **filtered_kwargs,
                    )
                else:
                    raise ValueError(
                        "Cluster client has no nodes - cannot create health check client"
                    )
            else:
                raise TypeError(f"Unsupported client type: {type(database.client)}")
            self._clients[db_id] = client

        return client

    async def close(self) -> None:
        """Close all health check clients."""
        close_tasks = [
            asyncio.create_task(client.aclose()) for client in self._clients.values()
        ]

        if close_tasks:
            await asyncio.gather(*close_tasks, return_exceptions=True)

        self._clients.clear()

    @abstractmethod
    async def _execute(self, health_check: HealthCheck, database) -> bool:
        """

View on GitHub (pinned to 6a6b581b48)