Weaviate: Fix to be compatible with python 3.8 (#1349)

use run_in_executor instead of to_thread to run the weaviate client ops
in async because to_thread is not available in python 3.8

python/semantic_kernel/connectors/memory/weaviate/weaviate_memory_store.py

@@ -133,14 +133,20 @@ class WeaviateMemoryStore(MemoryStoreBase):
     async def create_collection_async(self, collection_name: str) -> None:
         schema = SCHEMA.copy()
         schema["class"] = collection_name
-        await asyncio.to_thread(self.client.schema.create_class, schema)
+        await asyncio.get_running_loop().run_in_executor(
+            None, self.client.schema.create_class, schema
+        )
 
     async def get_collections_async(self) -> List[str]:
-        schemas = await asyncio.to_thread(self.client.schema.get)
+        schemas = await asyncio.get_running_loop().run_in_executor(
+            None, self.client.schema.get
+        )
         return [schema["class"] for schema in schemas["classes"]]
 
     async def delete_collection_async(self, collection_name: str) -> bool:
-        await asyncio.to_thread(self.client.schema.delete_class, collection_name)
+        await asyncio.get_running_loop().run_in_executor(
+            None, self.client.schema.delete_class, collection_name
+        )
 
     async def does_collection_exist_async(self, collection_name: str) -> bool:
         collections = await self.get_collections_async()
@@ -152,12 +158,13 @@ class WeaviateMemoryStore(MemoryStoreBase):
         vector = weaviate_record.pop("vector", None)
         weaviate_id = weaviate.util.generate_uuid5(weaviate_record, collection_name)
 
-        return await asyncio.to_thread(
+        return await asyncio.get_running_loop().run_in_executor(
+            None,
             self.client.data_object.create,
-            data_object=weaviate_record,
-            uuid=weaviate_id,
-            vector=vector,
-            class_name=collection_name,
+            weaviate_record,
+            collection_name,
+            weaviate_id,
+            vector,
         )
 
     async def upsert_batch_async(
@@ -182,7 +189,9 @@ class WeaviateMemoryStore(MemoryStoreBase):
 
             return results
 
-        return await asyncio.to_thread(_upsert_batch_inner)
+        return await asyncio.get_running_loop().run_in_executor(
+            None, _upsert_batch_inner
+        )
 
     async def get_async(
         self, collection_name: str, key: str, with_embedding: bool
@@ -196,7 +205,9 @@ class WeaviateMemoryStore(MemoryStoreBase):
     ) -> List[MemoryRecord]:
         queries = self._build_multi_get_query(collection_name, keys, with_embedding)
 
-        results = await asyncio.to_thread(self.client.query.multi_get(queries).do)
+        results = await asyncio.get_running_loop().run_in_executor(
+            None, self.client.query.multi_get(queries).do
+        )
 
         get_dict = results.get("data", {}).get("Get", {})
 
@@ -254,10 +265,8 @@ class WeaviateMemoryStore(MemoryStoreBase):
                 "valueString": key,
             }
 
-            await asyncio.to_thread(
-                self.client.batch.delete_objects,
-                class_name=collection_name,
-                where=where,
+            await asyncio.get_running_loop().run_in_executor(
+                None, self.client.batch.delete_objects, collection_name, where
             )
 
     async def get_nearest_matches_async(
@@ -284,7 +293,7 @@ class WeaviateMemoryStore(MemoryStoreBase):
             .with_limit(limit)
         )
 
-        results = await asyncio.to_thread(query.do)
+        results = await asyncio.get_running_loop().run_in_executor(None, query.do)
 
         get_dict = results.get("data", {}).get("Get", {})
 

python/tests/integration/connectors/memory/test_weaviate_memory_store.py

@@ -1,12 +1,13 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import sys
+import time
 
 import numpy as np
 import numpy.testing as npt
 import pytest
 
-from semantic_kernel.connectors.memory import weaviate_memory_store
+from semantic_kernel.connectors.memory.weaviate import weaviate_memory_store
 from semantic_kernel.memory.memory_record import MemoryRecord
 
 if not sys.platform.startswith("linux"):
@@ -67,10 +68,25 @@ def documents():
 
 @pytest.fixture
 def memory_store():
+    max_attempts = 5  # the number of retry attempts
+    delay = 30  # delay in seconds between each attempt
+
     config = weaviate_memory_store.WeaviateConfig(use_embed=True)
-    store = weaviate_memory_store.WeaviateMemoryStore(config)
-    store.client.schema.delete_all()
+    for attempt in range(max_attempts):
+        try:
+            store = weaviate_memory_store.WeaviateMemoryStore(config)
+            store.client.schema.delete_all()
+        except Exception:
+            if attempt < max_attempts - 1:  # it's not the final attempt
+                time.sleep(delay)  # wait before retrying
+                continue  # go to the next attempt
+            else:  # it's the final attempt
+                raise  # re-raise the last exception
+        else:
+            break  # successful attempt, get out of the loop
+
     yield store
+
     store.client.schema.delete_all()