Support redis-py 8

.github/workflows/test.yml

@@ -75,7 +75,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
-        redis-py-version: ["6.x", "7.x"]
+        redis-py-version: ["6.x", "7.x", "8.x"]
         redis-image: ["redis:8.2", "redis:8.4", "redis:latest"]
     steps:
       - name: Check out repository
@@ -112,6 +112,7 @@ jobs:
           case "$REDIS_PY_VERSION" in
             "6.x") spec="redis>=6,<7" ;;
             "7.x") spec="redis>=7,<8" ;;
+            "8.x") spec="redis>=8.1,<9" ;;
             *)
               echo "::error title=Unhandled redis-py-version::Matrix value '${REDIS_PY_VERSION}' has no install rule -- add a case branch in .github/workflows/test.yml."
               exit 1

pyproject.toml

@@ -25,7 +25,9 @@ classifiers = [
 dependencies = [
     "numpy>=1.26.0,<3",
     "pyyaml>=5.4,<7.0",
-    "redis>=5.0,<8.0",
+    # 6.3.0 first accepts SVS-VAMANA. redis-py 8.0.0 returns empty RESP3
+    # search results (redis-py #4107), so 8.1.0 is the supported 8.x floor.
+    "redis>=6.3.0,!=8.0.0,<9.0",
     "pydantic>=2,<3",
     "tenacity>=8.2.2",
     "ml-dtypes>=0.4.0,<1.0.0",

redisvl/cli/index.py

@@ -13,8 +13,12 @@
 )
 from redisvl.exceptions import RedisSearchError
 from redisvl.index import SearchIndex
-from redisvl.redis.connection import RedisConnectionFactory
-from redisvl.redis.utils import convert_bytes, make_dict
+from redisvl.redis.connection import (
+    RedisConnectionFactory,
+    normalize_index_definition,
+    normalize_index_fields,
+)
+from redisvl.redis.utils import convert_bytes
 from redisvl.schema.schema import IndexSchema
 from redisvl.utils.log import get_logger
 
@@ -275,27 +279,11 @@ def _connect_to_index(self, args: Namespace) -> SearchIndex:
 
 def _index_info_for_json(index_info: dict) -> dict:
     """Build the JSON payload from the same fields shown in table mode."""
-    definition_src = index_info.get("index_definition")
-    if isinstance(definition_src, list):
-        definition = convert_bytes(make_dict(definition_src))
-    elif isinstance(definition_src, tuple):
-        definition = convert_bytes(make_dict(list(definition_src)))
-    elif isinstance(definition_src, dict):
-        definition = convert_bytes(dict(definition_src))
-    else:
-        definition = {}
-    attributes = index_info.get("attributes", [])
+    definition = normalize_index_definition(index_info)
+    attributes = normalize_index_fields(index_info)
     index_fields = []
 
-    for attrs in attributes:
-        if isinstance(attrs, list):
-            attr = convert_bytes(make_dict(attrs))
-        elif isinstance(attrs, tuple):
-            attr = convert_bytes(make_dict(list(attrs)))
-        elif isinstance(attrs, dict):
-            attr = convert_bytes(dict(attrs))
-        else:
-            attr = {}
+    for attr in attributes:
         field = {
             "name": attr.get("identifier"),
             "attribute": attr.get("attribute"),
@@ -325,8 +313,8 @@ def _index_info_for_json(index_info: dict) -> dict:
 
 def _display_in_table(index_info):
     print("\n")
-    attributes = index_info.get("attributes", [])
-    definition = make_dict(index_info.get("index_definition"))
+    attributes = normalize_index_fields(index_info)
+    definition = normalize_index_definition(index_info)
     index_info = [
         index_info.get("index_name"),
         definition.get("key_type"),
@@ -370,18 +358,14 @@ def print_table_edge(length, col_width, start, mid, stop):
         "Type",
     ]
 
-    for attrs in attributes:
-        attr = make_dict(attrs)
-
+    for attr in attributes:
         values = [attr.get("identifier"), attr.get("attribute"), attr.get("type")]
-        if len(attrs) > 5:
-            options = make_dict(attrs)
-            for k, v in options.items():
-                if k not in ["identifier", "attribute", "type"]:
-                    headers.append("Field Option")
-                    headers.append("Option Value")
-                    values.append(k)
-                    values.append(v)
+        for k, v in attr.items():
+            if k not in ["identifier", "attribute", "type"]:
+                headers.append("Field Option")
+                headers.append("Option Value")
+                values.append(k)
+                values.append(v)
         attr_values.append(values)
 
     # Display the attributes in tabular format

redisvl/extensions/cache/embeddings/embeddings.py

@@ -1,5 +1,6 @@
 """Embeddings cache implementation for RedisVL."""
 
+from collections.abc import Mapping
 from typing import Any, Iterable
 
 from redisvl.extensions.cache.base import BaseCache
@@ -112,7 +113,9 @@ def _prepare_entry_data(
         )
         return key, entry.to_dict()
 
-    def _process_cache_data(self, data: dict[str, Any] | None) -> dict[str, Any] | None:
+    def _process_cache_data(
+        self, data: Mapping[bytes | str, Any] | None
+    ) -> dict[str, Any] | None:
         """Process Redis hash data into a cache entry response.
 
         Args:
@@ -124,7 +127,7 @@ def _process_cache_data(self, data: dict[str, Any] | None) -> dict[str, Any] | N
         if not data:
             return None
 
-        cache_hit = CacheEntry(**convert_bytes(data))
+        cache_hit = CacheEntry(**convert_bytes(dict(data)))
         return cache_hit.model_dump(exclude_none=True)
 
     def _should_warn_for_async_only(self) -> bool:

redisvl/index/index.py

@@ -47,21 +47,9 @@
     from redis.commands.search.result import Result
     from redisvl.query.query import BaseQuery
 
-from redis import __version__ as redis_version
 from redis.client import NEVER_DECODE
 from redis.commands.search.aggregation import AggregateRequest, Cursor
-
-from redisvl.utils.redis_protocol import get_protocol_version
-
-# Redis 5.x compatibility (6 fixed the import path)
-if redis_version.startswith("5"):
-    from redis.commands.search.indexDefinition import (
-        IndexDefinition,  # type: ignore[import-untyped]
-    )
-else:
-    from redis.commands.search.index_definition import (
-        IndexDefinition,  # type: ignore[no-redef]
-    )
+from redis.commands.search.index_definition import IndexDefinition
 
 # Need Result outside TYPE_CHECKING for cast
 from redis.commands.search.result import Result
@@ -101,9 +89,43 @@
     VectorIndexAlgorithm,
 )
 from redisvl.utils.log import get_logger
+from redisvl.utils.redis_protocol import get_protocol_version
 
 logger = get_logger(__name__)
 
+
+def _parse_batch_search_result(
+    search: Any, result: Any, query: Any, duration: float
+) -> Result:
+    """Parse a pipelined FT.SEARCH response across supported redis-py versions."""
+    parsed_result = search._parse_results(  # type: ignore
+        "FT.SEARCH", result, query=query, duration=duration
+    )
+    if not isinstance(parsed_result, dict):
+        return parsed_result
+
+    resp3_parser = getattr(search, "_parse_search_resp3", None)
+    if resp3_parser is not None:
+        return resp3_parser(parsed_result, query=query, duration=duration)
+
+    # redis-py 6.x returns the raw RESP3 map from _parse_results. Convert it to
+    # the RESP2 shape expected by its private _parse_search callback.
+    def value(mapping: dict[Any, Any], key: str, default: Any = None) -> Any:
+        return mapping.get(key, mapping.get(key.encode(), default))
+
+    response: list[Any] = [value(parsed_result, "total_results", 0)]
+    for document in value(parsed_result, "results", []):
+        response.append(value(document, "id", ""))
+        if query._with_scores:
+            response.append(value(document, "score", 0))
+        if query._with_payloads:
+            response.append(value(document, "payload"))
+        if not query._no_content:
+            fields = value(document, "extra_attributes", {})
+            response.append([item for pair in fields.items() for item in pair])
+    return search._parse_search(response, query=query, duration=duration)
+
+
 _HYBRID_SEARCH_ERROR_MESSAGE = "Hybrid search is not available in this version of redis-py. Please upgrade to redis-py >= 7.1.0."
 
 
@@ -1799,10 +1821,8 @@ def batch_search(
 
                 for j, query_results in enumerate(results):
                     _built_query = batch_built_queries[j]
-                    parsed_result = search._parse_search(  # type: ignore
-                        query_results,
-                        query=_built_query,
-                        duration=duration,
+                    parsed_result = _parse_batch_search_result(
+                        search, query_results, _built_query, duration
                     )
                     # Return a parsed Result object for each query
                     all_results.append(parsed_result)
@@ -2998,10 +3018,8 @@ async def batch_search(
 
                 for j, query_results in enumerate(results):
                     _built_query = batch_built_queries[j]
-                    parsed_result = search._parse_search(  # type: ignore
-                        query_results,
-                        query=_built_query,
-                        duration=duration,
+                    parsed_result = _parse_batch_search_result(
+                        search, query_results, _built_query, duration
                     )
                     # Return a parsed Result object for each query
                     all_results.append(parsed_result)

redisvl/index/storage.py

@@ -1,24 +1,12 @@
+import json
 from collections.abc import Collection
 from typing import Any, Callable, Iterable
 
 from pydantic import BaseModel, ValidationError
-from redis import __version__ as redis_version
-
-# Add imports for Pipeline types
 from redis.asyncio.client import Pipeline as AsyncPipeline
 from redis.asyncio.cluster import ClusterPipeline as AsyncClusterPipeline
-
-# Redis 5.x compatibility (6 fixed the import path)
-if redis_version.startswith("5"):
-    from redis.commands.search.indexDefinition import (  # type: ignore[import-untyped]
-        IndexType,
-    )
-else:
-    from redis.commands.search.index_definition import (  # type: ignore[no-redef]
-        IndexType,
-    )
-
-import json
+from redis.commands.search.index_definition import IndexType
+from redis.typing import EncodableT
 
 from redisvl.exceptions import SchemaValidationError
 from redisvl.redis.utils import convert_bytes
@@ -602,6 +590,19 @@ class HashStorage(BaseStorage):
     type: IndexType = IndexType.HASH
     """Hash data type for the index"""
 
+    @staticmethod
+    def _mapping(obj: dict[str, Any]) -> dict[EncodableT, EncodableT]:
+        """Build a redis-py encodable hash mapping."""
+        mapping: dict[EncodableT, EncodableT] = {}
+        for key, value in obj.items():
+            if not isinstance(value, (bytes, bytearray, memoryview, str, int, float)):
+                raise TypeError(
+                    f"Hash field {key!r} has unsupported value type "
+                    f"{type(value).__name__}"
+                )
+            mapping[key] = value
+        return mapping
+
     @staticmethod
     def _set(client: RedisClientOrPipeline, key: str, obj: dict[str, Any]):
         """Synchronously set a hash value in Redis for the given key.
@@ -611,7 +612,7 @@ def _set(client: RedisClientOrPipeline, key: str, obj: dict[str, Any]):
             key (str): The key under which to store the hash.
             obj (Dict[str, Any]): The hash to store in Redis.
         """
-        client.hset(name=key, mapping=obj)
+        client.hset(name=key, mapping=HashStorage._mapping(obj))
 
     @staticmethod
     async def _aset(client: AsyncRedisClientOrPipeline, key: str, obj: dict[str, Any]):
@@ -622,10 +623,11 @@ async def _aset(client: AsyncRedisClientOrPipeline, key: str, obj: dict[str, Any
             key (str): The key under which to store the hash.
             obj (Dict[str, Any]): The hash to store in Redis.
         """
+        mapping = HashStorage._mapping(obj)
         if isinstance(client, (AsyncPipeline, AsyncClusterPipeline)):
-            client.hset(name=key, mapping=obj)  # type: ignore
+            client.hset(name=key, mapping=mapping)
         else:
-            await client.hset(name=key, mapping=obj)  # type: ignore
+            await client.hset(name=key, mapping=mapping)
 
     @staticmethod
     def _get(client: SyncRedisClient, key: str) -> dict[str, Any]:

redisvl/mcp/config.py

@@ -433,7 +433,11 @@ def inspected_schema_from_index_info(
         incomplete on older Redis versions. MCP needs those field identities to survive
         so schema overrides can patch the missing attrs during startup.
         """
-        from redisvl.redis.connection import convert_index_info_to_schema
+        from redisvl.redis.connection import (
+            convert_index_info_to_schema,
+            normalize_index_definition,
+            normalize_index_fields,
+        )
 
         schema_dict = convert_index_info_to_schema(index_info)
         discovered_fields = {
@@ -442,18 +446,23 @@ def inspected_schema_from_index_info(
             if isinstance(field, dict) and "name" in field
         }
 
-        storage_type = index_info["index_definition"][1].lower()
-        for raw_field in index_info.get("attributes", []):
-            name = raw_field[1] if storage_type == "hash" else raw_field[3]
+        definition = normalize_index_definition(index_info)
+        storage_type = definition["key_type"].lower()
+        for raw_field in normalize_index_fields(index_info):
+            name = (
+                raw_field["identifier"]
+                if storage_type == "hash"
+                else raw_field["attribute"]
+            )
             if name in discovered_fields:
                 continue
 
             field = {
                 "name": name,
-                "type": str(raw_field[5]).lower(),
+                "type": str(raw_field["type"]).lower(),
             }
             if storage_type == "json":
-                field["path"] = raw_field[1]
+                field["path"] = raw_field["identifier"]
 
             # Keep discovered field identity even when FT.INFO omitted attrs.
             schema_dict.setdefault("fields", []).append(field)