Merge branch 'master' into temp-disable-aa

Author: petyaslavova

redis/__init__.py

@@ -1,6 +1,7 @@
 from redis import asyncio  # noqa
 from redis.backoff import default_backoff
 from redis.client import Redis, StrictRedis
+from redis.driver_info import DriverInfo
 from redis.cluster import RedisCluster
 from redis.connection import (
     BlockingConnectionPool,
@@ -63,6 +64,7 @@ def int_or_str(value):
     "CredentialProvider",
     "CrossSlotTransactionError",
     "DataError",
+    "DriverInfo",
     "from_url",
     "default_backoff",
     "InvalidPipelineStack",

redis/asyncio/client.py

@@ -53,6 +53,7 @@
     list_or_args,
 )
 from redis.credentials import CredentialProvider
+from redis.driver_info import DriverInfo, resolve_driver_info
 from redis.event import (
     AfterPooledConnectionsInstantiationEvent,
     AfterPubSubConnectionInstantiationEvent,
@@ -74,7 +75,6 @@
     _set_info_logger,
     deprecated_args,
     deprecated_function,
-    get_lib_version,
     safe_str,
     str_if_bytes,
     truncate_text,
@@ -214,6 +214,11 @@ def from_pool(
         reason="TimeoutError is included by default.",
         version="6.0.0",
     )
+    @deprecated_args(
+        args_to_warn=["lib_name", "lib_version"],
+        reason="Use 'driver_info' parameter instead. "
+        "lib_name and lib_version will be removed in a future version.",
+    )
     def __init__(
         self,
         *,
@@ -252,8 +257,9 @@ def __init__(
         single_connection_client: bool = False,
         health_check_interval: int = 0,
         client_name: Optional[str] = None,
-        lib_name: Optional[str] = "redis-py",
-        lib_version: Optional[str] = get_lib_version(),
+        lib_name: Optional[str] = None,
+        lib_version: Optional[str] = None,
+        driver_info: Optional["DriverInfo"] = None,
         username: Optional[str] = None,
         auto_close_connection_pool: Optional[bool] = None,
         redis_connect_func=None,
@@ -306,6 +312,12 @@ def __init__(
             # Create internal connection pool, expected to be closed by Redis instance
             if not retry_on_error:
                 retry_on_error = []
+
+            # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
+            computed_driver_info = resolve_driver_info(
+                driver_info, lib_name, lib_version
+            )
+
             kwargs = {
                 "db": db,
                 "username": username,
@@ -320,8 +332,7 @@ def __init__(
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
                 "client_name": client_name,
-                "lib_name": lib_name,
-                "lib_version": lib_version,
+                "driver_info": computed_driver_info,
                 "redis_connect_func": redis_connect_func,
                 "protocol": protocol,
             }

redis/asyncio/connection.py

@@ -38,6 +38,7 @@
     VerifyFlags = None
 
 from ..auth.token import TokenInterface
+from ..driver_info import DriverInfo, resolve_driver_info
 from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
 from ..utils import deprecated_args, format_error_message
 
@@ -63,7 +64,7 @@
     TimeoutError,
 )
 from redis.typing import EncodableT
-from redis.utils import HIREDIS_AVAILABLE, get_lib_version, str_if_bytes
+from redis.utils import HIREDIS_AVAILABLE, str_if_bytes
 
 from .._parsers import (
     BaseParser,
@@ -137,6 +138,11 @@ class AbstractConnection:
         "__dict__",
     )
 
+    @deprecated_args(
+        args_to_warn=["lib_name", "lib_version"],
+        reason="Use 'driver_info' parameter instead. "
+        "lib_name and lib_version will be removed in a future version.",
+    )
     def __init__(
         self,
         *,
@@ -153,8 +159,9 @@ def __init__(
         socket_read_size: int = 65536,
         health_check_interval: float = 0,
         client_name: Optional[str] = None,
-        lib_name: Optional[str] = "redis-py",
-        lib_version: Optional[str] = get_lib_version(),
+        lib_name: Optional[str] = None,
+        lib_version: Optional[str] = None,
+        driver_info: Optional[DriverInfo] = None,
         username: Optional[str] = None,
         retry: Optional[Retry] = None,
         redis_connect_func: Optional[ConnectCallbackT] = None,
@@ -163,6 +170,20 @@ def __init__(
         protocol: Optional[int] = 2,
         event_dispatcher: Optional[EventDispatcher] = None,
     ):
+        """
+        Initialize a new async Connection.
+
+        Parameters
+        ----------
+        driver_info : DriverInfo, optional
+            Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
+            are ignored. If not provided, a DriverInfo will be created from lib_name
+            and lib_version (or defaults if those are also None).
+        lib_name : str, optional
+            **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
+        lib_version : str, optional
+            **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
+        """
         if (username or password) and credential_provider is not None:
             raise DataError(
                 "'username' and 'password' cannot be passed along with 'credential_"
@@ -176,8 +197,10 @@ def __init__(
             self._event_dispatcher = event_dispatcher
         self.db = db
         self.client_name = client_name
-        self.lib_name = lib_name
-        self.lib_version = lib_version
+
+        # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
+        self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
+
         self.credential_provider = credential_provider
         self.password = password
         self.username = username
@@ -452,29 +475,36 @@ async def on_connect_check_health(self, check_health: bool = True) -> None:
             if str_if_bytes(await self.read_response()) != "OK":
                 raise ConnectionError("Error setting client name")
 
-        # set the library name and version, pipeline for lower startup latency
-        if self.lib_name:
+        # Set the library name and version from driver_info, pipeline for lower startup latency
+        lib_name_sent = False
+        lib_version_sent = False
+
+        if self.driver_info and self.driver_info.formatted_name:
             await self.send_command(
                 "CLIENT",
                 "SETINFO",
                 "LIB-NAME",
-                self.lib_name,
+                self.driver_info.formatted_name,
                 check_health=check_health,
             )
-        if self.lib_version:
+            lib_name_sent = True
+
+        if self.driver_info and self.driver_info.lib_version:
             await self.send_command(
                 "CLIENT",
                 "SETINFO",
                 "LIB-VER",
-                self.lib_version,
+                self.driver_info.lib_version,
                 check_health=check_health,
             )
+            lib_version_sent = True
+
         # if a database is specified, switch to it. Also pipeline this
         if self.db:
             await self.send_command("SELECT", self.db, check_health=check_health)
 
         # read responses from pipeline
-        for _ in (sent for sent in (self.lib_name, self.lib_version) if sent):
+        for _ in range(sum([lib_name_sent, lib_version_sent])):
             try:
                 await self.read_response()
             except ResponseError:
@@ -1215,16 +1245,17 @@ def can_get_connection(self) -> bool:
         version="5.3.0",
     )
     async def get_connection(self, command_name=None, *keys, **options):
+        """Get a connected connection from the pool"""
         async with self._lock:
-            """Get a connected connection from the pool"""
             connection = self.get_available_connection()
-            try:
-                await self.ensure_connection(connection)
-            except BaseException:
-                await self.release(connection)
-                raise
 
-        return connection
+        # We now perform the connection check outside of the lock.
+        try:
+            await self.ensure_connection(connection)
+            return connection
+        except BaseException:
+            await self.release(connection)
+            raise
 
     def get_available_connection(self):
         """Get a connection from the pool, without making sure it is connected"""

redis/client.py

@@ -40,6 +40,7 @@
     UnixDomainSocketConnection,
 )
 from redis.credentials import CredentialProvider
+from redis.driver_info import DriverInfo, resolve_driver_info
 from redis.event import (
     AfterPooledConnectionsInstantiationEvent,
     AfterPubSubConnectionInstantiationEvent,
@@ -63,7 +64,6 @@
 from redis.utils import (
     _set_info_logger,
     deprecated_args,
-    get_lib_version,
     safe_str,
     str_if_bytes,
     truncate_text,
@@ -199,6 +199,11 @@ def from_pool(
         reason="TimeoutError is included by default.",
         version="6.0.0",
     )
+    @deprecated_args(
+        args_to_warn=["lib_name", "lib_version"],
+        reason="Use 'driver_info' parameter instead. "
+        "lib_name and lib_version will be removed in a future version.",
+    )
     def __init__(
         self,
         host: str = "localhost",
@@ -240,8 +245,9 @@ def __init__(
         single_connection_client: bool = False,
         health_check_interval: int = 0,
         client_name: Optional[str] = None,
-        lib_name: Optional[str] = "redis-py",
-        lib_version: Optional[str] = get_lib_version(),
+        lib_name: Optional[str] = None,
+        lib_version: Optional[str] = None,
+        driver_info: Optional["DriverInfo"] = None,
         username: Optional[str] = None,
         redis_connect_func: Optional[Callable[[], None]] = None,
         credential_provider: Optional[CredentialProvider] = None,
@@ -280,6 +286,15 @@ def __init__(
         decode_responses:
             if `True`, the response will be decoded to utf-8.
             Argument is ignored when connection_pool is provided.
+        driver_info:
+            Optional DriverInfo object to identify upstream libraries.
+            If provided, lib_name and lib_version are ignored.
+            If not provided, a DriverInfo will be created from lib_name and lib_version.
+            Argument is ignored when connection_pool is provided.
+        lib_name:
+            **Deprecated.** Use driver_info instead. Library name for CLIENT SETINFO.
+        lib_version:
+            **Deprecated.** Use driver_info instead. Library version for CLIENT SETINFO.
         maint_notifications_config:
             configuration the pool to support maintenance notifications - see
             `redis.maint_notifications.MaintNotificationsConfig` for details.
@@ -296,6 +311,12 @@ def __init__(
         if not connection_pool:
             if not retry_on_error:
                 retry_on_error = []
+
+            # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
+            computed_driver_info = resolve_driver_info(
+                driver_info, lib_name, lib_version
+            )
+
             kwargs = {
                 "db": db,
                 "username": username,
@@ -309,8 +330,7 @@ def __init__(
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
                 "client_name": client_name,
-                "lib_name": lib_name,
-                "lib_version": lib_version,
+                "driver_info": computed_driver_info,
                 "redis_connect_func": redis_connect_func,
                 "credential_provider": credential_provider,
                 "protocol": protocol,
@@ -1009,10 +1029,22 @@ def is_health_check_response(self, response) -> bool:
         If there are no subscriptions redis responds to PING command with a
         bulk response, instead of a multi-bulk with "pong" and the response.
         """
-        return response in [
-            self.health_check_response,  # If there was a subscription
-            self.health_check_response_b,  # If there wasn't
-        ]
+        if self.encoder.decode_responses:
+            return (
+                response
+                in [
+                    self.health_check_response,  # If there is a subscription
+                    self.HEALTH_CHECK_MESSAGE,  # If there are no subscriptions and decode_responses=True
+                ]
+            )
+        else:
+            return (
+                response
+                in [
+                    self.health_check_response,  # If there is a subscription
+                    self.health_check_response_b,  # If there isn't a subscription and decode_responses=False
+                ]
+            )
 
     def check_health(self) -> None:
         conn = self.connection

redis/cluster.py

@@ -2208,7 +2208,8 @@ def _sharded_message_generator(self):
 
     def _pubsubs_generator(self):
         while True:
-            yield from self.node_pubsub_mapping.values()
+            current_nodes = list(self.node_pubsub_mapping.values())
+            yield from current_nodes
 
     def get_sharded_message(
         self, ignore_subscribe_messages=False, timeout=0.0, target_node=None