Merge branch 'master' into master

Author: JimNero009

.github/workflows/integration.yaml

@@ -74,7 +74,7 @@ jobs:
       max-parallel: 15
       fail-fast: false
       matrix:
-        redis-version: ['8.0-M04-pre', '${{ needs.redis_version.outputs.CURRENT }}', '7.2.7', '6.2.17']
+        redis-version: ['8.0-M05-pre', '${{ needs.redis_version.outputs.CURRENT }}', '7.2.7', '6.2.17']
         python-version: ['3.8', '3.13']
         parser-backend: ['plain']
         event-loop: ['asyncio']

redis/asyncio/client.py

@@ -2,7 +2,6 @@
 import copy
 import inspect
 import re
-import ssl
 import warnings
 from typing import (
     TYPE_CHECKING,
@@ -72,13 +71,19 @@
 from redis.typing import ChannelT, EncodableT, KeyT
 from redis.utils import (
     HIREDIS_AVAILABLE,
+    SSL_AVAILABLE,
     _set_info_logger,
     deprecated_function,
     get_lib_version,
     safe_str,
     str_if_bytes,
 )
 
+if TYPE_CHECKING and SSL_AVAILABLE:
+    from ssl import TLSVersion
+else:
+    TLSVersion = None
+
 PubSubHandler = Callable[[Dict[str, str]], Awaitable[None]]
 _KeyT = TypeVar("_KeyT", bound=KeyT)
 _ArgT = TypeVar("_ArgT", KeyT, EncodableT)
@@ -226,7 +231,7 @@ def __init__(
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_check_hostname: bool = False,
-        ssl_min_version: Optional[ssl.TLSVersion] = None,
+        ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
         max_connections: Optional[int] = None,
         single_connection_client: bool = False,

redis/asyncio/cluster.py

@@ -2,7 +2,6 @@
 import collections
 import random
 import socket
-import ssl
 import warnings
 from typing import (
     Any,
@@ -39,6 +38,7 @@
     SLOT_ID,
     AbstractRedisCluster,
     LoadBalancer,
+    LoadBalancingStrategy,
     block_pipeline_command,
     get_node_name,
     parse_cluster_slots,
@@ -64,7 +64,19 @@
     TryAgainError,
 )
 from redis.typing import AnyKeyT, EncodableT, KeyT
-from redis.utils import deprecated_function, get_lib_version, safe_str, str_if_bytes
+from redis.utils import (
+    SSL_AVAILABLE,
+    deprecated_args,
+    deprecated_function,
+    get_lib_version,
+    safe_str,
+    str_if_bytes,
+)
+
+if SSL_AVAILABLE:
+    from ssl import TLSVersion
+else:
+    TLSVersion = None
 
 TargetNodesT = TypeVar(
     "TargetNodesT", str, "ClusterNode", List["ClusterNode"], Dict[Any, "ClusterNode"]
@@ -111,9 +123,15 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         | See:
           https://redis.io/docs/manual/scaling/#redis-cluster-configuration-parameters
     :param read_from_replicas:
-        | Enable read from replicas in READONLY mode. You can read possibly stale data.
+        | @deprecated - please use load_balancing_strategy instead
+        | Enable read from replicas in READONLY mode.
           When set to true, read commands will be assigned between the primary and
           its replications in a Round-Robin manner.
+          The data read from replicas is eventually consistent with the data in primary nodes.
+    :param load_balancing_strategy:
+        | Enable read from replicas in READONLY mode and defines the load balancing
+          strategy that will be used for cluster node selection.
+          The data read from replicas is eventually consistent with the data in primary nodes.
     :param reinitialize_steps:
         | Specifies the number of MOVED errors that need to occur before reinitializing
           the whole cluster topology. If a MOVED error occurs and the cluster does not
@@ -206,6 +224,11 @@ def from_url(cls, url: str, **kwargs: Any) -> "RedisCluster":
         "result_callbacks",
     )
 
+    @deprecated_args(
+        args_to_warn=["read_from_replicas"],
+        reason="Please configure the 'load_balancing_strategy' instead",
+        version="5.0.3",
+    )
     def __init__(
         self,
         host: Optional[str] = None,
@@ -214,6 +237,7 @@ def __init__(
         startup_nodes: Optional[List["ClusterNode"]] = None,
         require_full_coverage: bool = True,
         read_from_replicas: bool = False,
+        load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
         reinitialize_steps: int = 5,
         cluster_error_retry_attempts: int = 3,
         connection_error_retry_attempts: int = 3,
@@ -247,7 +271,7 @@ def __init__(
         ssl_certfile: Optional[str] = None,
         ssl_check_hostname: bool = False,
         ssl_keyfile: Optional[str] = None,
-        ssl_min_version: Optional[ssl.TLSVersion] = None,
+        ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
         protocol: Optional[int] = 2,
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
@@ -312,7 +336,7 @@ def __init__(
                 }
             )
 
-        if read_from_replicas:
+        if read_from_replicas or load_balancing_strategy:
             # Call our on_connect function to configure READONLY mode
             kwargs["redis_connect_func"] = self.on_connect
 
@@ -361,6 +385,7 @@ def __init__(
         )
         self.encoder = Encoder(encoding, encoding_errors, decode_responses)
         self.read_from_replicas = read_from_replicas
+        self.load_balancing_strategy = load_balancing_strategy
         self.reinitialize_steps = reinitialize_steps
         self.cluster_error_retry_attempts = cluster_error_retry_attempts
         self.connection_error_retry_attempts = connection_error_retry_attempts
@@ -579,6 +604,7 @@ async def _determine_nodes(
             self.nodes_manager.get_node_from_slot(
                 await self._determine_slot(command, *args),
                 self.read_from_replicas and command in READ_COMMANDS,
+                self.load_balancing_strategy if command in READ_COMMANDS else None,
             )
         ]
 
@@ -759,7 +785,11 @@ async def _execute_command(
                     # refresh the target node
                     slot = await self._determine_slot(*args)
                     target_node = self.nodes_manager.get_node_from_slot(
-                        slot, self.read_from_replicas and args[0] in READ_COMMANDS
+                        slot,
+                        self.read_from_replicas and args[0] in READ_COMMANDS,
+                        self.load_balancing_strategy
+                        if args[0] in READ_COMMANDS
+                        else None,
                     )
                     moved = False
 
@@ -1221,17 +1251,23 @@ def _update_moved_slots(self) -> None:
         self._moved_exception = None
 
     def get_node_from_slot(
-        self, slot: int, read_from_replicas: bool = False
+        self,
+        slot: int,
+        read_from_replicas: bool = False,
+        load_balancing_strategy=None,
     ) -> "ClusterNode":
         if self._moved_exception:
             self._update_moved_slots()
 
+        if read_from_replicas is True and load_balancing_strategy is None:
+            load_balancing_strategy = LoadBalancingStrategy.ROUND_ROBIN
+
         try:
-            if read_from_replicas:
-                # get the server index in a Round-Robin manner
+            if len(self.slots_cache[slot]) > 1 and load_balancing_strategy:
+                # get the server index using the strategy defined in load_balancing_strategy
                 primary_name = self.slots_cache[slot][0].name
                 node_idx = self.read_load_balancer.get_server_index(
-                    primary_name, len(self.slots_cache[slot])
+                    primary_name, len(self.slots_cache[slot]), load_balancing_strategy
                 )
                 return self.slots_cache[slot][node_idx]
             return self.slots_cache[slot][0]

redis/asyncio/connection.py

@@ -3,7 +3,6 @@
 import enum
 import inspect
 import socket
-import ssl
 import sys
 import warnings
 import weakref
@@ -27,6 +26,16 @@
 )
 from urllib.parse import ParseResult, parse_qs, unquote, urlparse
 
+from ..utils import SSL_AVAILABLE
+
+if SSL_AVAILABLE:
+    import ssl
+    from ssl import SSLContext, TLSVersion
+else:
+    ssl = None
+    TLSVersion = None
+    SSLContext = None
+
 from ..auth.token import TokenInterface
 from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
 from ..utils import deprecated_args, format_error_message
@@ -763,10 +772,13 @@ def __init__(
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_check_hostname: bool = False,
-        ssl_min_version: Optional[ssl.TLSVersion] = None,
+        ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
         **kwargs,
     ):
+        if not SSL_AVAILABLE:
+            raise RedisError("Python wasn't built with SSL support")
+
         self.ssl_context: RedisSSLContext = RedisSSLContext(
             keyfile=ssl_keyfile,
             certfile=ssl_certfile,
@@ -834,9 +846,12 @@ def __init__(
         ca_certs: Optional[str] = None,
         ca_data: Optional[str] = None,
         check_hostname: bool = False,
-        min_version: Optional[ssl.TLSVersion] = None,
+        min_version: Optional[TLSVersion] = None,
         ciphers: Optional[str] = None,
     ):
+        if not SSL_AVAILABLE:
+            raise RedisError("Python wasn't built with SSL support")
+
         self.keyfile = keyfile
         self.certfile = certfile
         if cert_reqs is None:
@@ -857,9 +872,9 @@ def __init__(
         self.check_hostname = check_hostname
         self.min_version = min_version
         self.ciphers = ciphers
-        self.context: Optional[ssl.SSLContext] = None
+        self.context: Optional[SSLContext] = None
 
-    def get(self) -> ssl.SSLContext:
+    def get(self) -> SSLContext:
         if not self.context:
             context = ssl.create_default_context()
             context.check_hostname = self.check_hostname
@@ -1142,7 +1157,7 @@ async def ensure_connection(self, connection: AbstractConnection):
         try:
             if await connection.can_read_destructive():
                 raise ConnectionError("Connection has data") from None
-        except (ConnectionError, OSError):
+        except (ConnectionError, TimeoutError, OSError):
             await connection.disconnect()
             await connection.connect()
             if await connection.can_read_destructive():