Fix index.fetch() bug (#120)

`index.fetch()` was accidentally hardcoded to always pull data from
hash, even if the schema declared JSON for data storage. Updated the
method to use the `Storage` class for handling the get operation and
updated the tests.

Author: tylerhutcherson

redisvl/index/index.py

@@ -524,7 +524,7 @@ def add_field(d):
             logger.exception("Error while loading data to Redis")
             raise
 
-    def fetch(self, id: str) -> Dict[str, Any]:
+    def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         """Fetch an object from Redis by id.
 
         The id is typically either a unique identifier,
@@ -538,7 +538,10 @@ def fetch(self, id: str) -> Dict[str, Any]:
         Returns:
             Dict[str, Any]: The fetched object.
         """
-        return convert_bytes(self._redis_client.hgetall(self.key(id)))  # type: ignore
+        obj = self._storage.get(self._redis_client, [self.key(id)])  # type: ignore
+        if obj:
+            return convert_bytes(obj[0])
+        return None
 
     @check_index_exists()
     def search(self, *args, **kwargs) -> "Result":
@@ -901,7 +904,7 @@ async def add_field(d):
             logger.exception("Error while loading data to Redis")
             raise
 
-    async def fetch(self, id: str) -> Dict[str, Any]:
+    async def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         """Asynchronously etch an object from Redis by id. The id is typically
         either a unique identifier, or derived from some domain-specific
         metadata combination (like a document id or chunk id).
@@ -913,7 +916,10 @@ async def fetch(self, id: str) -> Dict[str, Any]:
         Returns:
             Dict[str, Any]: The fetched object.
         """
-        return convert_bytes(await self._redis_client.hgetall(self.key(id)))  # type: ignore
+        obj = await self._storage.aget(self._redis_client, [self.key(id)])  # type: ignore
+        if obj:
+            return convert_bytes(obj[0])
+        return None
 
     @check_async_index_exists()
     async def search(self, *args, **kwargs) -> "Result":

tests/integration/test_flow.py

@@ -54,9 +54,11 @@ def hash_preprocess(item: dict) -> dict:
         return {**item, "user_embedding": array_to_buffer(item["user_embedding"])}
 
     if index.storage_type == StorageType.HASH:
-        index.load(sample_data, preprocess=hash_preprocess)
+        index.load(sample_data, preprocess=hash_preprocess, id_field="user")
     else:
-        index.load(sample_data)
+        index.load(sample_data, id_field="user")
+
+    assert index.fetch("john")
 
     return_fields = ["user", "age", "job", "credit_score"]
     query = VectorQuery(

tests/integration/test_flow_async.py

@@ -57,9 +57,11 @@ async def hash_preprocess(item: dict) -> dict:
         return {**item, "user_embedding": array_to_buffer(item["user_embedding"])}
 
     if index.storage_type == StorageType.HASH:
-        await index.load(sample_data, preprocess=hash_preprocess)
+        await index.load(sample_data, preprocess=hash_preprocess, id_field="user")
     else:
-        await index.load(sample_data)
+        await index.load(sample_data, id_field="user")
+
+    assert await index.fetch("john")
 
     # wait for async index to create
     time.sleep(1)