WIP

Author: abrookins

redisvl/extensions/llmcache/semantic.py

@@ -174,17 +174,16 @@ def _modify_schema(
 
     async def _get_async_index(self) -> AsyncSearchIndex:
         """Lazily construct the async search index class."""
+        # Construct async index if necessary
         if not self._aindex:
-            # Construct async index if necessary
-            self._aindex = AsyncSearchIndex(schema=self._index.schema)
-            # Connect Redis async client
             redis_client = self.redis_kwargs["redis_client"]
             redis_url = self.redis_kwargs["redis_url"]
             connection_kwargs = self.redis_kwargs["connection_kwargs"]
-            if redis_client is not None:
-                await self._aindex.set_client(redis_client)
-            elif redis_url:
-                await self._aindex.connect(redis_url, **connection_kwargs)  # type: ignore
+            
+            self._aindex = AsyncSearchIndex(schema=self._index.schema,
+                                            redis_client=redis_client,
+                                            redis_url=redis_url,
+                                            **connection_kwargs)
         return self._aindex
 
     @property
@@ -290,7 +289,8 @@ async def _async_refresh_ttl(self, key: str) -> None:
         """Async refresh the time-to-live for the specified key."""
         aindex = await self._get_async_index()
         if self._ttl:
-            await aindex.client.expire(key, self._ttl)  # type: ignore
+            client = await aindex.get_client()
+            await client.expire(key, self._ttl)  # type: ignore
 
     def _vectorize_prompt(self, prompt: Optional[str]) -> List[float]:
         """Converts a text prompt to its vector representation using the

redisvl/index/index.py

@@ -15,6 +15,9 @@
     Optional,
     Union,
 )
+import warnings
+
+from redisvl.utils.utils import deprecated_function
 
 if TYPE_CHECKING:
     from redis.commands.search.aggregation import AggregateResult
@@ -839,6 +842,7 @@ def __init__(
     async def disconnect(self):
         """Asynchronously disconnect and cleanup the underlying async redis connection."""
         if self._redis_client is not None:
+            print(self._redis_client)
             await self._redis_client.aclose()  # type: ignore
         self._redis_client = None
 
@@ -860,12 +864,13 @@ async def from_existing(
             redis_url (Optional[str]): The URL of the Redis server to
                 connect to.
         """
-        if redis_url:
-            redis_client = RedisConnectionFactory.connect(
-                redis_url=redis_url, use_async=True, **kwargs
-            )
-
-        if not redis_client:
+        if redis_client and redis_url:
+            raise ValueError("Cannot provide both redis_client and redis_url")
+        elif redis_url:
+            redis_client = RedisConnectionFactory.get_async_redis_connection(url=redis_url, **kwargs)
+        elif redis_client:
+            pass
+        else:
             raise ValueError(
                 "Must provide either a redis_url or redis_client to fetch Redis index info."
             )
@@ -888,14 +893,21 @@ async def from_existing(
         index_info = await cls._info(name, redis_client)
         schema_dict = convert_index_info_to_schema(index_info)
         schema = IndexSchema.from_dict(schema_dict)
-        index = cls(schema, **kwargs)
-        await index.set_client(redis_client)
-        return index
+        return cls(schema, redis_client=redis_client, **kwargs)
 
+    @deprecated_function("client", "Use await self.get_client()")
     @property
-    def client(self) -> Optional[aredis.Redis]:
+    def client(self) -> aredis.Redis:
         """The underlying redis-py client object."""
-        return self._redis_client
+        if self._redis_client is None:
+            if asyncio.current_task() is not None:
+                warnings.warn("Risk of deadlock! Use await self.get_client() if you are "
+                              "in an async context.")
+            redis_client = asyncio.run_coroutine_threadsafe(self.get_client(), asyncio.get_event_loop())
+            client = redis_client.result()
+        else:
+            client = self._redis_client
+        return client
 
     async def connect(self, redis_url: Optional[str] = None, **kwargs):
         """[DEPRECATED] Connect to a Redis instance. Use connection parameters in __init__."""
@@ -910,31 +922,19 @@ async def connect(self, redis_url: Optional[str] = None, **kwargs):
         )
         return await self.set_client(client)
 
+    @deprecated_function("set_client", "Pass connection parameters in __init__")
     async def set_client(self, redis_client: Optional[aredis.Redis]):
-        """[DEPRECATED] Manually set the Redis client to use with the search index.
-        This method is deprecated; please provide connection parameters in __init__.
         """
-        import warnings
-
-        warnings.warn(
-            "set_client() is deprecated; pass connection parameters in __init__",
-            DeprecationWarning,
-        )
-        return await self._set_client(redis_client)
-
-    async def _set_client(self, redis_client: Optional[redis.asyncio.Redis]):
-        """
-        Set the Redis client to use with the search index.
-
-        NOTE: Remove this method once the deprecation period is over.
+        [DEPRECATED] Manually set the Redis client to use with the search index.
+        This method is deprecated; please provide connection parameters in __init__.
         """
         if self._redis_client is not None:
             await self._redis_client.aclose()  # type: ignore
         async with self._lock:
             self._redis_client = redis_client
         return self
 
-    async def _get_client(self) -> aredis.Redis:
+    async def get_client(self) -> aredis.Redis:
         """Lazily instantiate and return the async Redis client."""
         if self._redis_client is None:
             async with self._lock:
@@ -976,7 +976,7 @@ async def create(self, overwrite: bool = False, drop: bool = False) -> None:
             # overwrite an index in Redis; drop associated data (clean slate)
             await index.create(overwrite=True, drop=True)
         """
-        client = await self._get_client()
+        client = await self.get_client()
         redis_fields = self.schema.redis_fields
 
         if not redis_fields:
@@ -1012,7 +1012,7 @@ async def delete(self, drop: bool = True):
         Raises:
             redis.exceptions.ResponseError: If the index does not exist.
         """
-        client = await self._get_client()
+        client = await self.get_client()
         try:
             await client.ft(self.schema.index.name).dropindex(delete_documents=drop)
         except Exception as e:
@@ -1025,7 +1025,7 @@ async def clear(self) -> int:
         Returns:
             int: Count of records deleted from Redis.
         """
-        client = await self._get_client()
+        client = await self.get_client()
         total_records_deleted: int = 0
 
         async for batch in self.paginate(
@@ -1046,7 +1046,7 @@ async def drop_keys(self, keys: Union[str, List[str]]) -> int:
         Returns:
             int: Count of records deleted from Redis.
         """
-        client = await self._get_client()
+        client = await self.get_client()
         if isinstance(keys, list):
             return await client.delete(*keys)
         else:
@@ -1110,7 +1110,7 @@ async def add_field(d):
             keys = await index.load(data, preprocess=add_field)
 
         """
-        client = await self._get_client()
+        client = await self.get_client()
         try:
             return await self._storage.awrite(
                 client,
@@ -1137,7 +1137,7 @@ async def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         Returns:
             Dict[str, Any]: The fetched object.
         """
-        client = await self._get_client()
+        client = await self.get_client()
         obj = await self._storage.aget(client, [self.key(id)])
         if obj:
             return convert_bytes(obj[0])
@@ -1153,10 +1153,9 @@ async def aggregate(self, *args, **kwargs) -> "AggregateResult":
         Returns:
             Result: Raw Redis aggregation results.
         """
-        client = await self._get_client()
+        client = await self.get_client()
         try:
-            # TODO: Typing
-            return await client.ft(self.schema.index.name).aggregate(*args, **kwargs)
+            return client.ft(self.schema.index.name).aggregate(*args, **kwargs)
         except Exception as e:
             raise RedisSearchError(f"Error while aggregating: {str(e)}") from e
 
@@ -1170,10 +1169,9 @@ async def search(self, *args, **kwargs) -> "Result":
         Returns:
             Result: Raw Redis search results.
         """
-        client = await self._get_client()
+        client = await self.get_client()
         try:
-            # TODO: Typing
-            return await client.ft(self.schema.index.name).search(*args, **kwargs)
+            return client.ft(self.schema.index.name).search(*args, **kwargs)
         except Exception as e:
             raise RedisSearchError(f"Error while searching: {str(e)}") from e
 
@@ -1264,7 +1262,7 @@ async def listall(self) -> List[str]:
         Returns:
             List[str]: The list of indices in the database.
         """
-        client: aredis.Redis = await self._get_client()
+        client: aredis.Redis = await self.get_client()
         return convert_bytes(await client.execute_command("FT._LIST"))
 
     async def exists(self) -> bool:
@@ -1285,7 +1283,7 @@ async def info(self, name: Optional[str] = None) -> Dict[str, Any]:
         Returns:
             dict: A dictionary containing the information about the index.
         """
-        client: aredis.Redis = await self._get_client()
+        client: aredis.Redis = await self.get_client()
         index_name = name or self.schema.index.name
         return await type(self)._info(index_name, client)
 

redisvl/utils/utils.py

@@ -52,7 +52,7 @@ def validate_vector_dims(v1: int, v2: int) -> None:
 
 
 def serialize(data: Dict[str, Any]) -> str:
-    """Serlize the input into a string."""
+    """Serialize the input into a string."""
     return json.dumps(data)
 
 
@@ -88,3 +88,27 @@ def inner(*args, **kwargs):
         return inner
 
     return wrapper
+
+
+def deprecated_function(name: Optional[str] = None, replacement: Optional[str] = None):
+    """
+    Decorator to mark a function as deprecated.
+
+    When the wrapped function is called, the decorator will log a deprecation
+    warning.
+    """
+    def decorator(func):
+        fn_name = name or func.__name__
+        warning_message = f"Function {fn_name} is deprecated and will be " \
+            "removed in the next major release."
+        if replacement:
+            warning_message += replacement
+
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            warn(warning_message, category=DeprecationWarning, stacklevel=3)
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator

tests/integration/test_async_search_index.py

@@ -137,7 +137,7 @@ async def test_search_index_redis_url(redis_url, index_schema):
     )
     assert async_index.client
 
-    async_index.disconnect()
+    await async_index.disconnect()
     assert async_index.client == None
 
 
@@ -155,7 +155,7 @@ async def test_search_index_set_client(async_client, client, async_index):
     assert async_index.client == async_client
     await async_index.set_client(client)
 
-    async_index.disconnect()
+    await async_index.disconnect()
     assert async_index.client == None
 
 

tests/unit/test_utils.py

@@ -7,7 +7,7 @@
     convert_bytes,
     make_dict,
 )
-from redisvl.utils.utils import deprecated_argument
+from redisvl.utils.utils import deprecated_argument, deprecated_function
 
 
 def test_even_number_of_elements():
@@ -237,3 +237,22 @@ async def test_func(dtype=None, vectorizer=None):
         with pytest.warns(DeprecationWarning):
             result = await test_func(dtype="float32")
         assert result == 1
+
+
+
+class TestDeprecatedFunction:
+    def test_deprecated_function_warning(self):
+        @deprecated_function("new_func", "Use new_func2")
+        def old_func():
+            pass
+
+        with pytest.warns(DeprecationWarning):
+            old_func()
+
+    def test_deprecated_function_warning_with_name(self):
+        @deprecated_function("new_func", "Use new_func2")
+        def old_func():
+            pass
+
+        with pytest.warns(DeprecationWarning):
+            old_func()