Merge branch 'master' of github.com:redis/redis-py into vv-multidb-healthcheck-refactor

Author: vladvildanov

.agent/sync_async_type_hints_overload_guide.md

@@ -45,6 +45,7 @@ jobs:
           ignore-vulns: |
             GHSA-w596-4wvx-j9j6  # subversion related git pull, dependency for pytest. There is no impact here.
             CVE-2026-26007 # dependency for entraid tests
+            CVE-2026-32597 # PyJWT does not validate the crit (Critical) Header Parameter defined in RFC 7515, this will be fixed in the next release
 
   lint:
     name: Code linters

.github/workflows/integration.yaml

@@ -13,6 +13,7 @@
     Dict,
     Iterable,
     List,
+    Literal,
     Mapping,
     MutableMapping,
     Optional,
@@ -129,6 +130,9 @@ class Redis(
     Connection object to talk to redis.
     """
 
+    # Type discrimination marker for @overload self-type pattern
+    _is_async_client: Literal[True] = True
+
     response_callbacks: MutableMapping[Union[str, bytes], ResponseCallbackT]
 
     @classmethod

redis/asyncio/client.py

@@ -16,6 +16,7 @@
     Dict,
     Generator,
     List,
+    Literal,
     Mapping,
     Optional,
     Set,
@@ -238,6 +239,9 @@ def from_url(cls, url: str, **kwargs: Any) -> "RedisCluster":
             kwargs["ssl"] = True
         return cls(**kwargs)
 
+    # Type discrimination marker for @overload self-type pattern
+    _is_async_client: Literal[True] = True
+
     __slots__ = (
         "_initialize",
         "_lock",
@@ -1976,7 +1980,14 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
         | Existing :class:`~.RedisCluster` client
     """
 
-    __slots__ = ("cluster_client", "_transaction", "_execution_strategy")
+    __slots__ = (
+        "cluster_client",
+        "_transaction",
+        "_execution_strategy",
+    )
+
+    # Type discrimination marker for @overload self-type pattern
+    _is_async_client: Literal[True] = True
 
     def __init__(
         self, client: RedisCluster, transaction: Optional[bool] = None

redis/asyncio/cluster.py

@@ -59,6 +59,7 @@
 
 from redis.asyncio.observability.recorder import (
     record_connection_closed,
+    record_connection_count,
     record_connection_create_time,
     record_connection_wait_time,
     record_error_count,
@@ -1308,18 +1309,91 @@ def __init__(
         if self._event_dispatcher is None:
             self._event_dispatcher = EventDispatcher()
 
+    # Keys that should be redacted in __repr__ to avoid exposing sensitive information
+    SENSITIVE_REPR_KEYS = frozenset(
+        {
+            "password",
+            "username",
+            "ssl_password",
+            "credential_provider",
+        }
+    )
+
     def __repr__(self):
-        conn_kwargs = ",".join([f"{k}={v}" for k, v in self.connection_kwargs.items()])
+        conn_kwargs = ",".join(
+            [
+                f"{k}={'<REDACTED>' if k in self.SENSITIVE_REPR_KEYS else v}"
+                for k, v in self.connection_kwargs.items()
+            ]
+        )
         return (
             f"<{self.__class__.__module__}.{self.__class__.__name__}"
             f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
             f"({conn_kwargs})>)>"
         )
 
     def reset(self):
+        # Record metrics for connections being removed before clearing
+        # (only if attributes exist - they won't during __init__)
+        if hasattr(self, "_available_connections") and hasattr(
+            self, "_in_use_connections"
+        ):
+            idle_count = len(self._available_connections)
+            in_use_count = len(self._in_use_connections)
+            if idle_count > 0 or in_use_count > 0:
+                pool_name = get_pool_name(self)
+                # Note: Using sync version since reset() is sync
+                from redis.observability.recorder import (
+                    record_connection_count as sync_record_connection_count,
+                )
+
+                if idle_count > 0:
+                    sync_record_connection_count(
+                        pool_name=pool_name,
+                        connection_state=ConnectionState.IDLE,
+                        counter=-idle_count,
+                    )
+                if in_use_count > 0:
+                    sync_record_connection_count(
+                        pool_name=pool_name,
+                        connection_state=ConnectionState.USED,
+                        counter=-in_use_count,
+                    )
+
         self._available_connections = []
         self._in_use_connections = weakref.WeakSet()
 
+    def __del__(self) -> None:
+        """Clean up connection pool and record metrics when garbage collected."""
+        try:
+            if not hasattr(self, "_available_connections") or not hasattr(
+                self, "_in_use_connections"
+            ):
+                return
+            idle_count = len(self._available_connections)
+            in_use_count = len(self._in_use_connections)
+            if idle_count > 0 or in_use_count > 0:
+                pool_name = get_pool_name(self)
+                # Note: Using sync version since __del__ is sync
+                from redis.observability.recorder import (
+                    record_connection_count as sync_record_connection_count,
+                )
+
+                if idle_count > 0:
+                    sync_record_connection_count(
+                        pool_name=pool_name,
+                        connection_state=ConnectionState.IDLE,
+                        counter=-idle_count,
+                    )
+                if in_use_count > 0:
+                    sync_record_connection_count(
+                        pool_name=pool_name,
+                        connection_state=ConnectionState.USED,
+                        counter=-in_use_count,
+                    )
+        except Exception:
+            pass
+
     def can_get_connection(self) -> bool:
         """Return True if a connection can be retrieved from the pool."""
         return (
@@ -1346,6 +1420,29 @@ async def get_connection(self, command_name=None, *keys, **options):
             )
             is_created = connections_after > connections_before
 
+        # Record state transition for observability
+        # This ensures counters stay balanced if ensure_connection() fails and release() is called
+        pool_name = get_pool_name(self)
+        if is_created:
+            # New connection created and acquired: just USED +1
+            await record_connection_count(
+                pool_name=pool_name,
+                connection_state=ConnectionState.USED,
+                counter=1,
+            )
+        else:
+            # Existing connection acquired from pool: IDLE -> USED
+            await record_connection_count(
+                pool_name=pool_name,
+                connection_state=ConnectionState.IDLE,
+                counter=-1,
+            )
+            await record_connection_count(
+                pool_name=pool_name,
+                connection_state=ConnectionState.USED,
+                counter=1,
+            )
+
         # We now perform the connection check outside of the lock.
         try:
             await self.ensure_connection(connection)
@@ -1383,6 +1480,8 @@ def get_encoder(self):
 
     def make_connection(self):
         """Create a new connection.  Can be overridden by child classes."""
+        # Note: We don't record IDLE here because async uses a sync make_connection
+        # but async record_connection_count. The recording is handled in get_connection.
         return self.connection_class(**self.connection_kwargs)
 
     async def ensure_connection(self, connection: AbstractConnection):
@@ -1406,6 +1505,7 @@ async def release(self, connection: AbstractConnection):
         # Connections should always be returned to the correct pool,
         # not doing so is an error that will cause an exception here.
         self._in_use_connections.remove(connection)
+
         if connection.should_reconnect():
             await connection.disconnect()
 
@@ -1414,6 +1514,19 @@ async def release(self, connection: AbstractConnection):
             AsyncAfterConnectionReleasedEvent(connection)
         )
 
+        # Record state transition: USED -> IDLE
+        pool_name = get_pool_name(self)
+        await record_connection_count(
+            pool_name=pool_name,
+            connection_state=ConnectionState.USED,
+            counter=-1,
+        )
+        await record_connection_count(
+            pool_name=pool_name,
+            connection_state=ConnectionState.IDLE,
+            counter=1,
+        )
+
     async def disconnect(self, inuse_connections: bool = True):
         """
         Disconnects connections in the pool
@@ -1432,6 +1545,7 @@ async def disconnect(self, inuse_connections: bool = True):
             *(connection.disconnect() for connection in connections),
             return_exceptions=True,
         )
+
         exc = next((r for r in resp if isinstance(r, BaseException)), None)
         if exc:
             raise exc

redis/asyncio/connection.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Any, Awaitable, Callable, List, Optional, Union
+from typing import Any, Awaitable, Callable, List, Literal, Optional, Union
 
 from redis.asyncio.client import PubSubHandler
 from redis.asyncio.multidb.command_executor import DefaultCommandExecutor
@@ -437,6 +437,8 @@ class Pipeline(AsyncRedisModuleCommands, AsyncCoreCommands):
     Pipeline implementation for multiple logical Redis databases.
     """
 
+    _is_async_client: Literal[True] = True
+
     def __init__(self, client: MultiDBClient):
         self._command_stack = []
         self._client = client

redis/asyncio/multidb/client.py

@@ -24,13 +24,14 @@
 from typing import TYPE_CHECKING, List, Optional
 
 from redis.observability.attributes import (
+    ConnectionState,
     GeoFailoverReason,
     PubSubDirection,
 )
 from redis.observability.metrics import CloseReason, RedisMetricsCollector
 from redis.observability.providers import get_observability_instance
 from redis.observability.registry import get_observables_registry_instance
-from redis.utils import str_if_bytes
+from redis.utils import deprecated_function, str_if_bytes
 
 if TYPE_CHECKING:
     from redis.asyncio.connection import ConnectionPool
@@ -168,6 +169,38 @@ async def record_connection_create_time(
         pass
 
 
+async def record_connection_count(
+    pool_name: str,
+    connection_state: ConnectionState,
+    counter: int = 1,
+) -> None:
+    """
+    Record a connection count change for a single state.
+
+    Args:
+        pool_name: Connection pool identifier
+        connection_state: State to update (IDLE or USED)
+        counter: Number to add (positive) or subtract (negative)
+    """
+    collector = _get_or_create_collector()
+    if collector is None:
+        return
+
+    try:
+        collector.record_connection_count(
+            pool_name=pool_name,
+            connection_state=connection_state,
+            counter=counter,
+        )
+    except Exception:
+        pass
+
+
+@deprecated_function(
+    reason="Connection count is now tracked via record_connection_count(). "
+    "This functionality will be removed in the next major version",
+    version="7.4.0",
+)
 async def init_connection_count() -> None:
     """
     Initialize observable gauge for connection count metric.
@@ -194,6 +227,11 @@ def observable_callback(__):
         pass
 
 
+@deprecated_function(
+    reason="Connection count is now tracked via record_connection_count(). "
+    "This functionality will be removed in the next major version",
+    version="7.4.0",
+)
 async def register_pools_connection_count(
     connection_pools: List["ConnectionPool"],
 ) -> None:
@@ -301,7 +339,7 @@ async def record_connection_relaxed_timeout(
     Record a connection timeout relaxation event.
 
     Args:
-        connection_name: Connection identifier
+        connection_name: Connection identifier (pool name)
         maint_notification: Maintenance notification type
         relaxed: True to count up (relaxed), False to count down (unrelaxed)
     """

redis/asyncio/observability/recorder.py

@@ -10,6 +10,7 @@
     Callable,
     Dict,
     List,
+    Literal,
     Mapping,
     Optional,
     Set,
@@ -151,6 +152,9 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     It is not safe to pass PubSub or Pipeline objects between threads.
     """
 
+    # Type discrimination marker for @overload self-type pattern
+    _is_async_client: Literal[False] = False
+
     @classmethod
     def from_url(cls, url: str, **kwargs) -> "Redis":
         """