fix: Handle cluster parameter in AsyncRedisCluster connections (#346)

AsyncRedisCluster.from_url() doesn't accept 'cluster' parameter but it
might be present in URLs or kwargs for compatibility. This fix strips
the cluster parameter before creating async Redis connections.

Changes:
- Add _strip_cluster_from_url_and_kwargs helper function
- Update _get_aredis_connection to strip cluster parameter
- Update get_async_redis_cluster_connection to strip cluster parameter
- Update deprecated get_async_redis_connection to handle both cluster types
- Strip cluster parameter for both AsyncRedis and AsyncRedisCluster

Author: bsbodden

redisvl/redis/connection.py

@@ -1,5 +1,6 @@
 import os
-from typing import Any, Dict, List, Optional, Type
+from typing import Any, Dict, List, Optional, Tuple, Type
+from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 from warnings import warn
 
 from redis import Redis, RedisCluster
@@ -19,6 +20,52 @@
 from redisvl.utils.utils import deprecated_function
 
 
+def _strip_cluster_from_url_and_kwargs(
+    url: str, **kwargs
+) -> Tuple[str, Dict[str, Any]]:
+    """Remove 'cluster' parameter from URL query string and kwargs.
+
+    AsyncRedisCluster doesn't accept 'cluster' parameter, but it might be
+    present in the URL or kwargs for compatibility with other Redis clients.
+
+    Args:
+        url: Redis URL that might contain cluster parameter
+        **kwargs: Keyword arguments that might contain cluster parameter
+
+    Returns:
+        Tuple of (cleaned_url, cleaned_kwargs)
+    """
+    # Parse the URL
+    parsed = urlparse(url)
+
+    # Parse query parameters
+    query_params = parse_qs(parsed.query)
+
+    # Remove 'cluster' parameter if present
+    query_params.pop("cluster", None)
+
+    # Reconstruct the query string
+    new_query = urlencode(query_params, doseq=True)
+
+    # Reconstruct the URL
+    cleaned_url = urlunparse(
+        (
+            parsed.scheme,
+            parsed.netloc,
+            parsed.path,
+            parsed.params,
+            new_query,
+            parsed.fragment,
+        )
+    )
+
+    # Remove 'cluster' from kwargs if present
+    cleaned_kwargs = kwargs.copy()
+    cleaned_kwargs.pop("cluster", None)
+
+    return cleaned_url, cleaned_kwargs
+
+
 def compare_versions(version1: str, version2: str):
     """
     Compare two Redis version strings numerically.
@@ -300,9 +347,17 @@ async def _get_aredis_connection(
         url = url or get_address_from_env()
 
         if is_cluster_url(url, **kwargs):
-            client = AsyncRedisCluster.from_url(url, **kwargs)
+            # Strip 'cluster' parameter as AsyncRedisCluster doesn't accept it
+            cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(
+                url, **kwargs
+            )
+            client = AsyncRedisCluster.from_url(cleaned_url, **cleaned_kwargs)
         else:
-            client = AsyncRedis.from_url(url, **kwargs)
+            # Also strip cluster parameter for AsyncRedis to avoid connection issues
+            cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(
+                url, **kwargs
+            )
+            client = AsyncRedis.from_url(cleaned_url, **cleaned_kwargs)
 
         # Module validation removed - operations will fail naturally if modules are missing
         # Set client library name only
@@ -340,7 +395,20 @@ def get_async_redis_connection(
             DeprecationWarning,
         )
         url = url or get_address_from_env()
-        return AsyncRedis.from_url(url, **kwargs)
+
+        # Handle both cluster and non-cluster URLs
+        if is_cluster_url(url, **kwargs):
+            # Strip 'cluster' parameter as AsyncRedisCluster doesn't accept it
+            cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(
+                url, **kwargs
+            )
+            return AsyncRedisCluster.from_url(cleaned_url, **cleaned_kwargs)
+        else:
+            # Also strip cluster parameter for AsyncRedis to avoid connection issues
+            cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(
+                url, **kwargs
+            )
+            return AsyncRedis.from_url(cleaned_url, **cleaned_kwargs)
 
     @staticmethod
     def get_redis_cluster_connection(
@@ -358,7 +426,9 @@ def get_async_redis_cluster_connection(
     ) -> AsyncRedisCluster:
         """Creates and returns an asynchronous Redis client for a Redis cluster."""
         url = redis_url or get_address_from_env()
-        return AsyncRedisCluster.from_url(url, **kwargs)
+        # Strip 'cluster' parameter as AsyncRedisCluster doesn't accept it
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url, **kwargs)
+        return AsyncRedisCluster.from_url(cleaned_url, **cleaned_kwargs)
 
     @staticmethod
     def sync_to_async_redis(

tests/integration/test_async_cluster_connection.py

@@ -0,0 +1,85 @@
+"""Integration tests for AsyncRedisCluster connection with cluster parameter (issue #346)."""
+
+import pytest
+from redis.asyncio import Redis as AsyncRedis
+from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
+
+from redisvl.redis.connection import RedisConnectionFactory
+
+
+@pytest.mark.asyncio
+class TestAsyncClusterConnection:
+    """Test AsyncRedisCluster connections with cluster parameter."""
+
+    async def test_get_aredis_connection_with_cluster_url(self, redis_url):
+        """Test _get_aredis_connection handles cluster parameter in URL."""
+        # Add cluster=true to the URL (even though it's not actually a cluster)
+        # This simulates the issue where cluster=true is passed but not accepted
+        cluster_url = (
+            f"{redis_url}?cluster=false"  # Use false since we don't have a real cluster
+        )
+
+        # This should not raise a TypeError
+        client = await RedisConnectionFactory._get_aredis_connection(cluster_url)
+
+        assert client is not None
+        assert isinstance(client, (AsyncRedis, AsyncRedisCluster))
+
+        await client.aclose()
+
+    async def test_get_aredis_connection_with_cluster_kwargs(self, redis_url):
+        """Test _get_aredis_connection handles cluster parameter in kwargs."""
+        # This should not raise a TypeError even with cluster in kwargs
+        client = await RedisConnectionFactory._get_aredis_connection(
+            redis_url, cluster=False  # Use false since we don't have a real cluster
+        )
+
+        assert client is not None
+        assert isinstance(client, (AsyncRedis, AsyncRedisCluster))
+
+        await client.aclose()
+
+    @pytest.mark.requires_cluster
+    async def test_get_async_redis_cluster_connection_with_params(
+        self, redis_cluster_url
+    ):
+        """Test get_async_redis_cluster_connection with cluster parameter."""
+        # Add cluster=true to the URL
+        cluster_url_with_param = f"{redis_cluster_url}?cluster=true"
+
+        # This should not raise a TypeError
+        client = RedisConnectionFactory.get_async_redis_cluster_connection(
+            cluster_url_with_param
+        )
+
+        assert client is not None
+        assert isinstance(client, AsyncRedisCluster)
+
+        await client.aclose()
+
+    @pytest.mark.requires_cluster
+    async def test_get_async_redis_cluster_connection_with_kwargs(
+        self, redis_cluster_url
+    ):
+        """Test get_async_redis_cluster_connection with cluster in kwargs."""
+        # This should not raise a TypeError
+        client = RedisConnectionFactory.get_async_redis_cluster_connection(
+            redis_cluster_url, cluster=True
+        )
+
+        assert client is not None
+        assert isinstance(client, AsyncRedisCluster)
+
+        await client.aclose()
+
+    def test_get_async_redis_connection_deprecated_with_cluster(self, redis_url):
+        """Test deprecated get_async_redis_connection handles cluster parameter."""
+        # Add cluster=false to the URL
+        cluster_url = f"{redis_url}?cluster=false"
+
+        with pytest.warns(DeprecationWarning):
+            # This should not raise a TypeError
+            client = RedisConnectionFactory.get_async_redis_connection(cluster_url)
+
+        assert client is not None
+        assert isinstance(client, (AsyncRedis, AsyncRedisCluster))

tests/unit/test_async_cluster_fix.py

@@ -0,0 +1,83 @@
+"""Unit tests for AsyncRedisCluster cluster parameter stripping fix (issue #346)."""
+
+import pytest
+
+from redisvl.redis.connection import _strip_cluster_from_url_and_kwargs
+
+
+class TestAsyncClusterParameterStripping:
+    """Test the helper function that strips cluster parameter from URLs and kwargs."""
+
+    def test_strip_cluster_from_url_with_cluster_true(self):
+        """Test stripping cluster=true from URL query string."""
+        url = "redis://localhost:7001?cluster=true"
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url)
+
+        assert cleaned_url == "redis://localhost:7001"
+        assert cleaned_kwargs == {}
+
+    def test_strip_cluster_from_url_with_other_params(self):
+        """Test stripping cluster parameter while preserving other parameters."""
+        url = (
+            "redis://localhost:7001?cluster=true&decode_responses=true&socket_timeout=5"
+        )
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(
+            url, some_kwarg="value"
+        )
+
+        assert "cluster" not in cleaned_url
+        assert "decode_responses=true" in cleaned_url
+        assert "socket_timeout=5" in cleaned_url
+        assert cleaned_kwargs == {"some_kwarg": "value"}
+
+    def test_strip_cluster_from_kwargs(self):
+        """Test stripping cluster parameter from kwargs."""
+        url = "redis://localhost:7001"
+        kwargs = {"cluster": True, "decode_responses": True}
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url, **kwargs)
+
+        assert cleaned_url == "redis://localhost:7001"
+        assert "cluster" not in cleaned_kwargs
+        assert cleaned_kwargs == {"decode_responses": True}
+
+    def test_strip_cluster_from_both_url_and_kwargs(self):
+        """Test stripping cluster parameter from both URL and kwargs."""
+        url = "redis://localhost:7001?cluster=true"
+        kwargs = {"cluster": True, "socket_timeout": 5}
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url, **kwargs)
+
+        assert cleaned_url == "redis://localhost:7001"
+        assert cleaned_kwargs == {"socket_timeout": 5}
+
+    def test_no_cluster_parameter_unchanged(self):
+        """Test that URLs and kwargs without cluster parameter remain unchanged."""
+        url = "redis://localhost:7001?decode_responses=true"
+        kwargs = {"socket_timeout": 5}
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url, **kwargs)
+
+        assert cleaned_url == "redis://localhost:7001?decode_responses=true"
+        assert cleaned_kwargs == {"socket_timeout": 5}
+
+    def test_empty_url_query_and_kwargs(self):
+        """Test handling of URL without query string and empty kwargs."""
+        url = "redis://localhost:7001"
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url)
+
+        assert cleaned_url == "redis://localhost:7001"
+        assert cleaned_kwargs == {}
+
+    def test_complex_url_with_auth_and_db(self):
+        """Test complex URL with authentication and database selection."""
+        url = "redis://user:password@localhost:7001/0?cluster=true&socket_timeout=5"
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url)
+
+        assert cleaned_url == "redis://user:password@localhost:7001/0?socket_timeout=5"
+        assert cleaned_kwargs == {}
+
+    def test_cluster_false_also_stripped(self):
+        """Test that cluster=false is also stripped (any cluster param should be removed)."""
+        url = "redis://localhost:7001?cluster=false"
+        cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url)
+
+        assert cleaned_url == "redis://localhost:7001"
+        assert cleaned_kwargs == {}