Handle new messages in SSB 1.35 API

.github/workflows/main.yaml

@@ -26,6 +26,7 @@ env:
   WEAVIATE_132: 1.32.16
   WEAVIATE_133: 1.33.4
   WEAVIATE_134: 1.34.0
+  WEAVIATE_135: 1.35.0-dev-8d38bb2.amd64
 
 jobs:
   lint-and-format:
@@ -304,7 +305,8 @@ jobs:
           $WEAVIATE_131,
           $WEAVIATE_132,
           $WEAVIATE_133,
-          $WEAVIATE_134
+          $WEAVIATE_134,
+          $WEAVIATE_135
         ]
     steps:
       - name: Checkout

profiling/conftest.py

@@ -61,6 +61,10 @@ def _factory(
             headers=headers,
             additional_config=AdditionalConfig(timeout=(60, 120)),  # for image tests
         )
+        # client_fixture = weaviate.connect_to_weaviate_cloud(
+        #     cluster_url="flnyoj61teuw1mxfwf1fsa.c0.europe-west3.gcp.weaviate.cloud",
+        #     auth_credentials=weaviate.auth.Auth.api_key("QnVtdnlnM2RYeUh3NVlFNF82V3pqVEtoYnloMlo0MHV2R2hYMU9BUFFsR3cvUUlkUG9CTFRiQXNjam1nPV92MjAw"),
+        # )
         client_fixture.collections.delete(name_fixture)
         if integration_config is not None:
             client_fixture.integrations.configure(integration_config)

profiling/test_sphere.py

@@ -28,7 +28,7 @@ def test_sphere(collection_factory: CollectionFactory) -> None:
     start = time.time()
 
     import_objects = 1000000
-    with collection.batch.dynamic() as batch:
+    with collection.batch.experimental() as batch:
         with open(sphere_file) as jsonl_file:
             for i, jsonl in enumerate(jsonl_file):
                 if i == import_objects or batch.number_errors > 10:
@@ -46,7 +46,7 @@ def test_sphere(collection_factory: CollectionFactory) -> None:
                     vector=json_parsed["vector"],
                 )
                 if i % 1000 == 0:
-                    print(f"Imported {len(collection)} objects")
+                    print(f"Imported {len(collection)} objects after processing {i} lines")
     assert len(collection.batch.failed_objects) == 0
     assert len(collection) == import_objects
     print(f"Imported {import_objects} objects in {time.time() - start}")

weaviate/connect/v4.py

@@ -391,7 +391,7 @@ def _open_connections_rest(
             async def get_oidc() -> None:
                 async with self._make_client("async") as client:
                     try:
-                        response = await client.get(oidc_url)
+                        response = await client.get(oidc_url, timeout=self.timeout_config.init)
                     except Exception as e:
                         raise WeaviateConnectionError(
                             f"Error: {e}. \nIs Weaviate running and reachable at {self.url}?"
@@ -406,7 +406,7 @@ async def get_oidc() -> None:
 
         with self._make_client("sync") as client:
             try:
-                response = client.get(oidc_url)
+                response = client.get(oidc_url, timeout=self.timeout_config.init)
             except Exception as e:
                 raise WeaviateConnectionError(
                     f"Error: {e}. \nIs Weaviate running and reachable at {self.url}?"
@@ -1013,7 +1013,7 @@ def grpc_batch_stream(
                 raise InsufficientPermissionsError(error)
             if error.code() == StatusCode.ABORTED:
                 raise _BatchStreamShutdownError()
-            raise WeaviateBatchStreamError(str(error.details()))
+            raise WeaviateBatchStreamError(f"{error.code()}({error.details()})")
 
     def grpc_batch_delete(
         self, request: batch_delete_pb2.BatchDeleteRequest

weaviate/exceptions.py

@@ -292,6 +292,15 @@ def __init__(self, message: str):
         self.message = message
 
 
+class WeaviateBatchFailedToReestablishStreamError(WeaviateBaseError):
+    """Is raised when the batch stream fails to re-establish within a timeout period."""
+
+    def __init__(self, message: str):
+        msg = f"""Batch stream failed to re-establish: {message}"""
+        super().__init__(msg)
+        self.message = message
+
+
 class WeaviateInsertInvalidPropertyError(WeaviateBaseError):
     """Is raised when inserting an invalid property."""