Dev/1.36

.github/workflows/main.yaml

@@ -27,6 +27,8 @@ env:
   WEAVIATE_133: 1.33.10
   WEAVIATE_134: 1.34.5
   WEAVIATE_135: 1.35.0
+  WEAVIATE_136: 1.36.0-rc.0
+
 
 jobs:
   lint-and-format:
@@ -154,11 +156,11 @@ jobs:
       fail-fast: false
       matrix:
         versions: [
-          { py: "3.10", weaviate: $WEAVIATE_132, grpc: "1.59.0"},
-          { py: "3.11", weaviate: $WEAVIATE_132, grpc: "1.66.0"},
-          { py: "3.12", weaviate: $WEAVIATE_132, grpc: "1.70.0"},
-          { py: "3.13", weaviate: $WEAVIATE_132, grpc: "1.72.1"},
-          { py: "3.14", weaviate: $WEAVIATE_132, grpc: "1.76.0"}
+          { py: "3.10", weaviate: $WEAVIATE_136, grpc: "1.59.0"},
+          { py: "3.11", weaviate: $WEAVIATE_136, grpc: "1.66.0"},
+          { py: "3.12", weaviate: $WEAVIATE_136, grpc: "1.70.0"},
+          { py: "3.13", weaviate: $WEAVIATE_136, grpc: "1.72.1"},
+          { py: "3.14", weaviate: $WEAVIATE_136, grpc: "1.76.0"}
         ]
         optional_dependencies: [false]
     steps:
@@ -209,11 +211,11 @@ jobs:
       fail-fast: false
       matrix:
         versions: [
-          { py: "3.10", weaviate: $WEAVIATE_132},
-          { py: "3.11", weaviate: $WEAVIATE_132},
-          { py: "3.12", weaviate: $WEAVIATE_132},
-          { py: "3.13", weaviate: $WEAVIATE_132},
-          { py: "3.14", weaviate: $WEAVIATE_132}
+          { py: "3.10", weaviate: $WEAVIATE_136},
+          { py: "3.11", weaviate: $WEAVIATE_136},
+          { py: "3.12", weaviate: $WEAVIATE_136},
+          { py: "3.13", weaviate: $WEAVIATE_136},
+          { py: "3.14", weaviate: $WEAVIATE_136}
         ]
         optional_dependencies: [false]
     steps:
@@ -305,8 +307,9 @@ jobs:
           $WEAVIATE_131,
           $WEAVIATE_132,
           $WEAVIATE_133,
-          $WEAVIATE_134
-          $WEAVIATE_135
+          $WEAVIATE_134,
+          $WEAVIATE_135,
+          $WEAVIATE_136
         ]
     steps:
       - name: Checkout

integration/test_backup_v4.py

@@ -497,13 +497,13 @@ def test_backup_and_restore_with_collection_and_config_1_24_x(
 
 
 @pytest.mark.parametrize("dynamic_backup_location", [False, True])
-def test_cancel_backup(
+def test_cancel_backup_create(
     client: weaviate.WeaviateClient,
     dynamic_backup_location: bool,
     tmp_path: pathlib.Path,
     request: SubRequest,
 ) -> None:
-    """Cancel backup without waiting."""
+    """Cancel backup create without waiting."""
     backup_id = unique_backup_id(request.node.name)
     if client._connection._weaviate_version.is_lower_than(1, 24, 25):
         pytest.skip("Cancel backups is only supported from 1.24.25")
@@ -543,6 +543,67 @@ def test_cancel_backup(
     assert status_resp.status == BackupStatus.CANCELED or status_resp.status == BackupStatus.SUCCESS
 
 
+@pytest.mark.parametrize("dynamic_backup_location", [False, True])
+def test_cancel_backup_restore(
+    client: weaviate.WeaviateClient,
+    dynamic_backup_location: bool,
+    tmp_path: pathlib.Path,
+    request: SubRequest,
+) -> None:
+    """Cancel backup restore without waiting."""
+    backup_id = unique_backup_id(request.node.name)
+    if client._connection._weaviate_version.is_lower_than(1, 36, 0):
+        pytest.skip("Cancel restores is only supported from 1.36.0")
+
+    backup_location: Optional[wvc.backup.BackupLocationType] = None
+    if dynamic_backup_location:
+        backup_location = wvc.backup.BackupLocation.FileSystem(path=str(tmp_path))
+
+    c_name = "Restore"
+    c = client.collections.create(
+        name=c_name, properties=[Property(name="name", data_type=DataType.TEXT)]
+    )
+    c.data.insert({"name": "test"})
+
+    resp = client.backup.create(
+        backup_id=backup_id,
+        backend=BACKEND,
+        backup_location=backup_location,
+        include_collections=[c_name],
+        wait_for_completion=True,
+    )
+    assert resp.status == BackupStatus.SUCCESS
+
+    client.collections.delete(c_name)
+
+    resp = client.backup.restore(
+        backup_id=backup_id,
+        backend=BACKEND,
+        backup_location=backup_location,
+        include_collections=[c_name],
+    )
+    assert resp.status == BackupStatus.STARTED
+
+    assert client.backup.cancel(
+        backup_id=backup_id, backend=BACKEND, backup_location=backup_location, operation="restore"
+    )
+
+    start = time.time()
+    while time.time() - start < 5:
+        status_resp = client.backup.get_restore_status(
+            backup_id=backup_id, backend=BACKEND, backup_location=backup_location
+        )
+        if status_resp.status == BackupStatus.CANCELED:
+            break
+        time.sleep(0.1)
+
+    status_resp = client.backup.get_restore_status(
+        backup_id=backup_id, backend=BACKEND, backup_location=backup_location
+    )
+    # there can be a race between the cancel and the backup completion
+    assert status_resp.status == BackupStatus.CANCELED or status_resp.status == BackupStatus.SUCCESS
+
+
 def test_backup_and_restore_with_roles_and_users(
     client_factory: ClientFactory, request: SubRequest
 ) -> None:

integration/test_batch_v4.py

@@ -4,6 +4,7 @@
 from typing import Callable, Generator, List, Optional, Protocol, Tuple
 
 import pytest
+import pytest_asyncio
 from _pytest.fixtures import SubRequest
 
 import weaviate
@@ -119,6 +120,53 @@ def _factory(
             client_fixture.close()
 
 
+class AsyncClientFactory(Protocol):
+    """Typing for fixture."""
+
+    async def __call__(
+        self, name: str = "", ports: Tuple[int, int] = (8080, 50051), multi_tenant: bool = False
+    ) -> Tuple[weaviate.WeaviateAsyncClient, str]:
+        """Typing for fixture."""
+        ...
+
+
+@pytest_asyncio.fixture
+async def async_client_factory(request: SubRequest):
+    name_fixtures: List[str] = []
+    client_fixture: Optional[weaviate.WeaviateAsyncClient] = None
+
+    async def _factory(
+        name: str = "", ports: Tuple[int, int] = (8080, 50051), multi_tenant: bool = False
+    ):
+        nonlocal client_fixture, name_fixtures  # noqa: F824
+        name_fixture = _sanitize_collection_name(request.node.name) + name
+        name_fixtures.append(name_fixture)
+        if client_fixture is None:
+            client_fixture = weaviate.use_async_with_local(grpc_port=ports[1], port=ports[0])
+            await client_fixture.connect()
+
+        if await client_fixture.collections.exists(name_fixture):
+            await client_fixture.collections.delete(name_fixture)
+
+        await client_fixture.collections.create(
+            name=name_fixture,
+            properties=[
+                Property(name="name", data_type=DataType.TEXT),
+                Property(name="age", data_type=DataType.INT),
+            ],
+            references=[ReferenceProperty(name="test", target_collection=name_fixture)],
+            multi_tenancy_config=Configure.multi_tenancy(multi_tenant),
+            vectorizer_config=Configure.Vectorizer.none(),
+        )
+        return client_fixture, name_fixture
+
+    try:
+        yield _factory
+    finally:
+        if client_fixture is not None:
+            await client_fixture.close()
+
+
 def test_add_objects_in_multiple_batches(client_factory: ClientFactory) -> None:
     client, name = client_factory()
     with client.batch.rate_limit(50) as batch:
@@ -365,16 +413,14 @@ def test_add_ref_batch_with_tenant(client_factory: ClientFactory) -> None:
 @pytest.mark.parametrize(
     "batching_method",
     [
-        # lambda client: client.batch.dynamic(),
-        # lambda client: client.batch.fixed_size(),
-        # lambda client: client.batch.rate_limit(9999),
-        lambda client: client.batch.experimental(concurrency=1),
+        lambda client: client.batch.dynamic(),
+        lambda client: client.batch.fixed_size(),
+        lambda client: client.batch.stream(concurrency=1),
     ],
     ids=[
-        # "test_add_ten_thousand_data_objects_dynamic",
-        # "test_add_ten_thousand_data_objects_fixed_size",
-        # "test_add_ten_thousand_data_objects_rate_limit",
-        "test_add_ten_thousand_data_objects_experimental",
+        "test_add_ten_thousand_data_objects_dynamic",
+        "test_add_ten_thousand_data_objects_fixed_size",
+        "test_add_ten_thousand_data_objects_stream",
     ],
 )
 def test_add_ten_thousand_data_objects(
@@ -385,10 +431,10 @@ def test_add_ten_thousand_data_objects(
     """Test adding ten thousand data objects."""
     client, name = client_factory()
     if (
-        request.node.callspec.id == "test_add_ten_thousand_data_objects_experimental"
-        and client._connection._weaviate_version.is_lower_than(1, 34, 0)
+        request.node.callspec.id == "test_add_ten_thousand_data_objects_stream"
+        and client._connection._weaviate_version.is_lower_than(1, 36, 0)
     ):
-        pytest.skip("Server-side batching not supported in Weaviate < 1.34.0")
+        pytest.skip("Server-side batching not supported in Weaviate < 1.36.0")
     nr_objects = 100000
     import time
 
@@ -434,9 +480,11 @@ def test_add_one_hundred_objects_and_references_between_all(
 ) -> None:
     """Test adding one hundred objects and references between all of them."""
     client, name = client_factory()
+    if client._connection._weaviate_version.is_lower_than(1, 36, 0):
+        pytest.skip("Server-side batching not supported in Weaviate < 1.36.0")
     nr_objects = 100
     uuids: List[UUID] = []
-    with client.batch.dynamic() as batch:
+    with client.batch.stream() as batch:
         for i in range(nr_objects):
             uuid_ = batch.add_object(
                 collection=name,
@@ -575,14 +623,12 @@ def test_add_1000_tenant_objects_with_async_indexing_and_wait_for_only_one(
     [
         lambda client: client.batch.dynamic(),
         lambda client: client.batch.fixed_size(),
-        lambda client: client.batch.rate_limit(1000),
-        lambda client: client.batch.experimental(),
+        lambda client: client.batch.stream(),
     ],
     ids=[
-        "test_add_one_hundred_objects_and_references_between_all_dynamic",
-        "test_add_one_hundred_objects_and_references_between_all_fixed_size",
-        "test_add_one_hundred_objects_and_references_between_all_rate_limit",
-        "test_add_one_hundred_objects_and_references_between_all_experimental",
+        "test_add_one_object_and_a_self_reference_dynamic",
+        "test_add_one_object_and_a_self_reference_fixed_size",
+        "test_add_one_object_and_a_self_reference_stream",
     ],
 )
 def test_add_one_object_and_a_self_reference(
@@ -593,11 +639,10 @@ def test_add_one_object_and_a_self_reference(
     """Test adding one object and a self reference."""
     client, name = client_factory()
     if (
-        request.node.callspec.id
-        == "test_add_one_hundred_objects_and_references_between_all_experimental"
-        and client._connection._weaviate_version.is_lower_than(1, 34, 0)
+        request.node.callspec.id == "test_add_one_object_and_a_self_reference_stream"
+        and client._connection._weaviate_version.is_lower_than(1, 36, 0)
     ):
-        pytest.skip("Server-side batching not supported in Weaviate < 1.34.0")
+        pytest.skip("Server-side batching not supported in Weaviate < 1.36.0")
     with batching_method(client) as batch:
         uuid = batch.add_object(collection=name, properties={})
         batch.add_reference(
@@ -768,3 +813,64 @@ def test_references_with_to_uuids(client_factory: ClientFactory) -> None:
 
     assert len(client.batch.failed_references) == 0, client.batch.failed_references
     client.collections.delete(["target", "source"])
+
+
+@pytest.mark.asyncio
+async def test_add_one_hundred_thousand_objects_async_client(
+    async_client_factory: AsyncClientFactory,
+) -> None:
+    """Test adding one hundred thousand data objects."""
+    client, name = await async_client_factory()
+    if client._connection._weaviate_version.is_lower_than(1, 36, 0):
+        pytest.skip("Server-side batching not supported in Weaviate < 1.36.0")
+    nr_objects = 100000
+    import time
+
+    start = time.time()
+    async with client.batch.stream(concurrency=1) as batch:
+        for i in range(nr_objects):
+            await batch.add_object(
+                collection=name,
+                properties={"name": "test" + str(i)},
+            )
+    end = time.time()
+    print(f"Time taken to add {nr_objects} objects: {end - start} seconds")
+    assert len(client.batch.results.objs.errors) == 0
+    assert len(client.batch.results.objs.all_responses) == nr_objects
+    assert len(client.batch.results.objs.uuids) == nr_objects
+    assert await client.collections.use(name).length() == nr_objects
+    assert client.batch.results.objs.has_errors is False
+    assert len(client.batch.failed_objects) == 0, [
+        obj.message for obj in client.batch.failed_objects
+    ]
+    await client.collections.delete(name)
+
+
+def test_add_one_hundred_thousand_objects_sync_client(
+    client_factory: ClientFactory,
+) -> None:
+    """Test adding one hundred thousand data objects."""
+    client, name = client_factory()
+    if client._connection._weaviate_version.is_lower_than(1, 36, 0):
+        pytest.skip("Server-side batching not supported in Weaviate < 1.36.0")
+    nr_objects = 100000
+    import time
+
+    start = time.time()
+    with client.batch.stream(concurrency=1) as batch:
+        for i in range(nr_objects):
+            batch.add_object(
+                collection=name,
+                properties={"name": "test" + str(i)},
+            )
+    end = time.time()
+    print(f"Time taken to add {nr_objects} objects: {end - start} seconds")
+    assert len(client.batch.results.objs.errors) == 0
+    assert len(client.batch.results.objs.all_responses) == nr_objects
+    assert len(client.batch.results.objs.uuids) == nr_objects
+    assert len(client.collections.use(name)) == nr_objects
+    assert client.batch.results.objs.has_errors is False
+    assert len(client.batch.failed_objects) == 0, [
+        obj.message for obj in client.batch.failed_objects
+    ]
+    client.collections.delete(name)