Enable creating an index "from_existing" (#174)

Since version 2.8.10 of the search module in Redis, `FT.INFO` now
reports the contents of the index and associated fields and all low
level attributes. Raw response from Redis:
```
 1) index_name
 2) user_simple
 3) index_options
 4) (empty array)
 5) index_definition
 6) 1) key_type
    2) HASH
    3) prefixes
    4) 1) user_simple_docs
    5) default_score
    6) "1"
 7) attributes
 8) 1) 1) identifier
       2) user
       3) attribute
       4) user
       5) type
       6) TAG
       7) SEPARATOR
       8) ,
    2) 1) identifier
       2) credit_score
       3) attribute
       4) credit_score
       5) type
       6) TAG
       7) SEPARATOR
       8) ,
    3) 1) identifier
       2) job
       3) attribute
       4) job
       5) type
       6) TEXT
       7) WEIGHT
       8) "1"
    4) 1) identifier
       2) age
       3) attribute
       4) age
       5) type
       6) NUMERIC
    5)  1) identifier
        2) user_embedding
        3) attribute
        4) user_embedding
        5) type
        6) VECTOR
        7) algorithm
        8) FLAT
        9) data_type
       10) FLOAT32
       11) dim
       12) (integer) 3
       13) distance_metric
       14) COSINE
 9) num_docs
10) "0"
11) max_doc_id
12) "0"
13) num_terms
14) "0"
15) num_records
16) "0"
17) inverted_sz_mb
18) "0"
19) vector_index_sz_mb
20) "0.00818634033203125"
21) total_inverted_index_blocks
22) "1813"
23) offset_vectors_sz_mb
24) "0"
25) doc_table_size_mb
26) "0"
27) sortable_values_size_mb
28) "0"
29) key_table_size_mb
30) "0"
31) geoshapes_sz_mb
32) "0"
33) records_per_doc_avg
34) "nan"
35) bytes_per_record_avg
36) "nan"
37) offsets_per_term_avg
38) "nan"
39) offset_bits_per_record_avg
40) "nan"
41) hash_indexing_failures
42) "0"
43) total_indexing_time
44) "0"
45) indexing
46) "0"
47) percent_indexed
48) "1"
49) number_of_uses
50) (integer) 2
51) cleaning
52) (integer) 0
53) gc_stats
54)  1) bytes_collected
     2) "0"
     3) total_ms_run
     4) "0"
     5) total_cycles
     6) "0"
     7) average_cycle_time_ms
     8) "nan"
     9) last_run_time_ms
    10) "0"
    11) gc_numeric_trees_missed
    12) "0"
    13) gc_blocks_denied
    14) "0"
55) cursor_stats
56) 1) global_idle
    2) (integer) 0
    3) global_total
    4) (integer) 0
    5) index_capacity
    6) (integer) 128
    7) index_total
    8) (integer) 0
57) dialect_stats
58) 1) dialect_1
    2) (integer) 0
    3) dialect_2
    4) (integer) 0
    5) dialect_3
    6) (integer) 0
    7) dialect_4
    8) (integer) 0
59) Index Errors
60) 1) indexing failures
    2) (integer) 0
    3) last indexing error
    4) N/A
    5) last indexing error key
    6) "N/A"
61) field statistics
62) 1) 1) identifier
       2) user
       3) attribute
       4) user
       5) Index Errors
       6) 1) indexing failures
          2) (integer) 0
          3) last indexing error
          4) N/A
          5) last indexing error key
          6) "N/A"
    2) 1) identifier
       2) credit_score
       3) attribute
       4) credit_score
       5) Index Errors
       6) 1) indexing failures
          2) (integer) 0
          3) last indexing error
          4) N/A
          5) last indexing error key
          6) "N/A"
    3) 1) identifier
       2) job
       3) attribute
       4) job
       5) Index Errors
       6) 1) indexing failures
          2) (integer) 0
          3) last indexing error
          4) N/A
          5) last indexing error key
          6) "N/A"
    4) 1) identifier
       2) age
       3) attribute
       4) age
       5) Index Errors
       6) 1) indexing failures
          2) (integer) 0
          3) last indexing error
          4) N/A
          5) last indexing error key
          6) "N/A"
    5) 1) identifier
       2) user_embedding
       3) attribute
       4) user_embedding
       5) Index Errors
       6) 1) indexing failures
          2) (integer) 0
          3) last indexing error
          4) N/A
          5) last indexing error key
          6) "N/A"
```

This enables the ability to "hydrate" a RedisVL `IndexSchema` class from
this output. This makes state management for index information MUCH
simpler. The caveat is that this capability is only available in newer
versions of redis/search. So we have to hide it behind a "Feature flag"
of sorts. This is directly needed in our integration clients like
LangChain and LlamaIndex too.

Author: tylerhutcherson

redisvl/index/index.py

@@ -24,7 +24,11 @@
 
 from redisvl.index.storage import HashStorage, JsonStorage
 from redisvl.query.query import BaseQuery, CountQuery, FilterQuery
-from redisvl.redis.connection import RedisConnectionFactory
+from redisvl.redis.connection import (
+    RedisConnectionFactory,
+    convert_index_info_to_schema,
+    validate_modules,
+)
 from redisvl.redis.utils import convert_bytes
 from redisvl.schema import IndexSchema, StorageType
 from redisvl.utils.log import get_logger
@@ -102,7 +106,7 @@ def decorator(func):
         @wraps(func)
         def wrapper(self, *args, **kwargs):
             if not self.exists():
-                raise ValueError(
+                raise RuntimeError(
                     f"Index has not been created. Must be created before calling {func.__name__}"
                 )
             return func(self, *args, **kwargs)
@@ -162,7 +166,6 @@ def __init__(
 
         self.schema = schema
 
-        # set custom lib name
         self._lib_name: Optional[str] = kwargs.pop("lib_name", None)
 
         # set up redis connection
@@ -317,6 +320,34 @@ class SearchIndex(BaseSearchIndex):
 
     """
 
+    @classmethod
+    def from_existing(
+        cls,
+        name: str,
+        redis_client: Optional[redis.Redis] = None,
+        redis_url: Optional[str] = None,
+        **kwargs,
+    ):
+        # Handle redis instance
+        if redis_url:
+            redis_client = RedisConnectionFactory.connect(
+                redis_url=redis_url, use_async=False, **kwargs
+            )
+        if not redis_client:
+            raise ValueError(
+                "Must provide either a redis_url or redis_client to fetch Redis index info."
+            )
+
+        # Validate modules
+        installed_modules = RedisConnectionFactory._get_modules(redis_client)
+        validate_modules(installed_modules, [{"name": "search", "ver": 20810}])
+
+        # Fetch index info and convert to schema
+        index_info = cls._info(name, redis_client)
+        schema_dict = convert_index_info_to_schema(index_info)
+        schema = IndexSchema.from_dict(schema_dict)
+        return cls(schema, redis_client, **kwargs)
+
     def connect(self, redis_url: Optional[str] = None, **kwargs):
         """Connect to a Redis instance using the provided `redis_url`, falling
         back to the `REDIS_URL` environment variable (if available).
@@ -653,22 +684,28 @@ def exists(self) -> bool:
         """
         return self.schema.index.name in self.listall()
 
+    @staticmethod
+    def _info(name: str, redis_client: redis.Redis) -> Dict[str, Any]:
+        """Run FT.INFO to fetch information about the index."""
+        try:
+            return convert_bytes(redis_client.ft(name).info())  # type: ignore
+        except:
+            logger.exception(f"Error while fetching {name} index info")
+            raise
+
     @check_index_exists()
-    def info(self) -> Dict[str, Any]:
+    def info(self, name: Optional[str] = None) -> Dict[str, Any]:
         """Get information about the index.
 
+        Args:
+            name (str, optional): Index name to fetch info about.
+                Defaults to None.
+
         Returns:
             dict: A dictionary containing the information about the index.
         """
-        try:
-            return convert_bytes(
-                self._redis_client.ft(self.schema.index.name).info()  # type: ignore
-            )
-        except:
-            logger.exception(
-                f"Error while fetching {self.schema.index.name} index info"
-            )
-            raise
+        index_name = name or self.schema.index.name
+        return self._info(index_name, self._redis_client)  # type: ignore
 
 
 class AsyncSearchIndex(BaseSearchIndex):
@@ -698,6 +735,36 @@ class AsyncSearchIndex(BaseSearchIndex):
 
     """
 
+    @classmethod
+    async def from_existing(
+        cls,
+        name: str,
+        redis_client: Optional[aredis.Redis] = None,
+        redis_url: Optional[str] = None,
+        **kwargs,
+    ):
+        if redis_url:
+            redis_client = RedisConnectionFactory.connect(
+                redis_url=redis_url, use_async=True, **kwargs
+            )
+
+        if not redis_client:
+            raise ValueError(
+                "Must provide either a redis_url or redis_client to fetch Redis index info."
+            )
+
+        # Validate modules
+        installed_modules = await RedisConnectionFactory._get_modules_async(
+            redis_client
+        )
+        validate_modules(installed_modules, [{"name": "search", "ver": 20810}])
+
+        # Fetch index info and convert to schema
+        index_info = await cls._info(name, redis_client)
+        schema_dict = convert_index_info_to_schema(index_info)
+        schema = IndexSchema.from_dict(schema_dict)
+        return cls(schema, redis_client, **kwargs)
+
     def connect(self, redis_url: Optional[str] = None, **kwargs):
         """Connect to a Redis instance using the provided `redis_url`, falling
         back to the `REDIS_URL` environment variable (if available).
@@ -1035,19 +1102,24 @@ async def exists(self) -> bool:
         """
         return self.schema.index.name in await self.listall()
 
+    @staticmethod
+    async def _info(name: str, redis_client: aredis.Redis) -> Dict[str, Any]:
+        try:
+            return convert_bytes(await redis_client.ft(name).info())  # type: ignore
+        except:
+            logger.exception(f"Error while fetching {name} index info")
+            raise
+
     @check_async_index_exists()
-    async def info(self) -> Dict[str, Any]:
+    async def info(self, name: Optional[str] = None) -> Dict[str, Any]:
         """Get information about the index.
 
+        Args:
+            name (str, optional): Index name to fetch info about.
+                Defaults to None.
+
         Returns:
             dict: A dictionary containing the information about the index.
         """
-        try:
-            return convert_bytes(
-                await self._redis_client.ft(self.schema.index.name).info()  # type: ignore
-            )
-        except:
-            logger.exception(
-                f"Error while fetching {self.schema.index.name} index info"
-            )
-            raise
+        index_name = name or self.schema.index.name
+        return await self._info(index_name, self._redis_client)  # type: ignore

redisvl/redis/connection.py

@@ -13,11 +13,16 @@
 )
 from redis.exceptions import ResponseError
 
-from redisvl.redis.constants import REDIS_REQUIRED_MODULES
+from redisvl.redis.constants import DEFAULT_REQUIRED_MODULES
 from redisvl.redis.utils import convert_bytes
 from redisvl.version import __version__
 
 
+def unpack_redis_modules(module_list: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Unpack a list of Redis modules pulled from the MODULES LIST command."""
+    return {module["name"]: module["ver"] for module in module_list}
+
+
 def get_address_from_env() -> str:
     """Get a redis connection from environment variables.
 
@@ -43,6 +48,82 @@ def make_lib_name(*args) -> str:
     return f"redis-py({custom_libs})"
 
 
+def convert_index_info_to_schema(index_info: Dict[str, Any]) -> Dict[str, Any]:
+    """Convert the output of FT.INFO into a schema-ready dictionary.
+
+    Args:
+        index_info (Dict[str, Any]): Output of the Redis FT.INFO command.
+
+    Returns:
+        Dict[str, Any]: Schema dictionary.
+    """
+    index_name = index_info["index_name"]
+    prefixes = index_info["index_definition"][3][0]
+    storage_type = index_info["index_definition"][1].lower()
+
+    index_fields = index_info["attributes"]
+
+    def parse_vector_attrs(attrs):
+        vector_attrs = {attrs[i].lower(): attrs[i + 1] for i in range(6, len(attrs), 2)}
+        vector_attrs["dims"] = int(vector_attrs.pop("dim"))
+        vector_attrs["distance_metric"] = vector_attrs.pop("distance_metric").lower()
+        vector_attrs["algorithm"] = vector_attrs.pop("algorithm").lower()
+        vector_attrs["datatype"] = vector_attrs.pop("data_type").lower()
+        return vector_attrs
+
+    def parse_attrs(attrs):
+        return {attrs[i].lower(): attrs[i + 1] for i in range(6, len(attrs), 2)}
+
+    schema_fields = []
+
+    for field_attrs in index_fields:
+        # parse field info
+        name = field_attrs[1] if storage_type == "hash" else field_attrs[3]
+        field = {"name": name, "type": field_attrs[5].lower()}
+        if storage_type == "json":
+            field["path"] = field_attrs[1]
+        # parse field attrs
+        if field_attrs[5] == "VECTOR":
+            field["attrs"] = parse_vector_attrs(field_attrs)
+        else:
+            field["attrs"] = parse_attrs(field_attrs)
+        # append field
+        schema_fields.append(field)
+
+    return {
+        "index": {"name": index_name, "prefix": prefixes, "storage_type": storage_type},
+        "fields": schema_fields,
+    }
+
+
+def validate_modules(
+    installed_modules: Dict[str, Any],
+    required_modules: Optional[List[Dict[str, Any]]] = None,
+) -> None:
+    """
+    Validates if required Redis modules are installed.
+
+    Args:
+        installed_modules: List of installed modules.
+        required_modules: List of required modules.
+
+    Raises:
+        ValueError: If required Redis modules are not installed.
+    """
+    required_modules = required_modules or DEFAULT_REQUIRED_MODULES
+
+    for required_module in required_modules:
+        if required_module["name"] in installed_modules:
+            installed_version = installed_modules[required_module["name"]]  # type: ignore
+            if int(installed_version) >= int(required_module["ver"]):  # type: ignore
+                return
+
+    raise ValueError(
+        f"Required Redis database module {required_module['name']} with version >= {required_module['ver']} not installed. "
+        "See Redis Stack documentation: https://redis.io/docs/stack/"
+    )
+
+
 class RedisConnectionFactory:
     """Builds connections to a Redis database, supporting both synchronous and
     asynchronous clients.
@@ -128,14 +209,14 @@ def get_async_redis_connection(url: Optional[str] = None, **kwargs) -> AsyncRedi
     def validate_redis(
         client: Union[Redis, AsyncRedis],
         lib_name: Optional[str] = None,
-        redis_required_modules: Optional[List[Dict[str, Any]]] = None,
+        required_modules: Optional[List[Dict[str, Any]]] = None,
     ) -> None:
         """Validates the Redis connection.
 
         Args:
             client (Redis or AsyncRedis): Redis client.
             lib_name (str): Library name to set on the Redis client.
-            redis_required_modules (List[Dict[str, Any]]): List of required modules and their versions.
+            required_modules (List[Dict[str, Any]]): List of required modules and their versions.
 
         Raises:
             ValueError: If required Redis modules are not installed.
@@ -145,18 +226,26 @@ def validate_redis(
                 RedisConnectionFactory._validate_async_redis,
                 client,
                 lib_name,
-                redis_required_modules,
+                required_modules,
             )
         else:
             RedisConnectionFactory._validate_sync_redis(
-                client, lib_name, redis_required_modules
+                client, lib_name, required_modules
             )
 
+    @staticmethod
+    def _get_modules(client: Redis) -> Dict[str, Any]:
+        return unpack_redis_modules(convert_bytes(client.module_list()))
+
+    @staticmethod
+    async def _get_modules_async(client: AsyncRedis) -> Dict[str, Any]:
+        return unpack_redis_modules(convert_bytes(await client.module_list()))
+
     @staticmethod
     def _validate_sync_redis(
         client: Redis,
         lib_name: Optional[str],
-        redis_required_modules: Optional[List[Dict[str, Any]]],
+        required_modules: Optional[List[Dict[str, Any]]],
     ) -> None:
         """Validates the sync client."""
         # Set client library name
@@ -168,16 +257,16 @@ def _validate_sync_redis(
             client.echo(_lib_name)
 
         # Get list of modules
-        modules_list = convert_bytes(client.module_list())
+        installed_modules = RedisConnectionFactory._get_modules(client)
 
         # Validate available modules
-        RedisConnectionFactory._validate_modules(modules_list, redis_required_modules)
+        validate_modules(installed_modules, required_modules)
 
     @staticmethod
     async def _validate_async_redis(
         client: AsyncRedis,
         lib_name: Optional[str],
-        redis_required_modules: Optional[List[Dict[str, Any]]],
+        required_modules: Optional[List[Dict[str, Any]]],
     ) -> None:
         """Validates the async client."""
         # Set client library name
@@ -189,10 +278,10 @@ async def _validate_async_redis(
             await client.echo(_lib_name)
 
         # Get list of modules
-        modules_list = convert_bytes(await client.module_list())
+        installed_modules = await RedisConnectionFactory._get_modules_async(client)
 
         # Validate available modules
-        RedisConnectionFactory._validate_modules(modules_list, redis_required_modules)
+        validate_modules(installed_modules, required_modules)
 
     @staticmethod
     def _run_async(coro, *args, **kwargs):
@@ -232,31 +321,3 @@ def _run_async(coro, *args, **kwargs):
             finally:
                 # Close the event loop to release resources
                 loop.close()
-
-    @staticmethod
-    def _validate_modules(
-        installed_modules, redis_required_modules: Optional[List[Dict[str, Any]]] = None
-    ) -> None:
-        """
-        Validates if required Redis modules are installed.
-
-        Args:
-            installed_modules: List of installed modules.
-            redis_required_modules: List of required modules.
-
-        Raises:
-            ValueError: If required Redis modules are not installed.
-        """
-        installed_modules = {module["name"]: module for module in installed_modules}
-        redis_required_modules = redis_required_modules or REDIS_REQUIRED_MODULES
-
-        for required_module in redis_required_modules:
-            if required_module["name"] in installed_modules:
-                installed_version = installed_modules[required_module["name"]]["ver"]
-                if int(installed_version) >= int(required_module["ver"]):  # type: ignore
-                    return
-
-        raise ValueError(
-            f"Required Redis database module {required_module['name']} with version >= {required_module['ver']} not installed. "
-            "Refer to Redis Stack documentation: https://redis.io/docs/stack/"
-        )

redisvl/redis/constants.py

@@ -1,5 +1,5 @@
 # required modules
-REDIS_REQUIRED_MODULES = [
+DEFAULT_REQUIRED_MODULES = [
     {"name": "search", "ver": 20600},
     {"name": "searchlight", "ver": 20600},
 ]