Dev/1.36

.github/workflows/main.yaml

@@ -22,11 +22,12 @@ env:
   WEAVIATE_128: 1.28.16
   WEAVIATE_129: 1.29.11
   WEAVIATE_130: 1.30.22
-  WEAVIATE_131: 1.31.20
-  WEAVIATE_132: 1.32.23
-  WEAVIATE_133: 1.33.10
-  WEAVIATE_134: 1.34.5
-  WEAVIATE_135: 1.35.0
+  WEAVIATE_131: 1.31.22
+  WEAVIATE_132: 1.32.26
+  WEAVIATE_133: 1.33.11
+  WEAVIATE_134: 1.34.8
+  WEAVIATE_135: 1.35.2
+  WEAVIATE_136: 1.36.0-dev-0bbf31a
 
 jobs:
   lint-and-format:
@@ -154,11 +155,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 +210,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 +306,9 @@ jobs:
           $WEAVIATE_131,
           $WEAVIATE_132,
           $WEAVIATE_133,
-          $WEAVIATE_134
-          $WEAVIATE_135
+          $WEAVIATE_134,
+          $WEAVIATE_135,
+          $WEAVIATE_136
         ]
     steps:
       - name: Checkout

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
 
@@ -575,14 +621,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 +637,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 +811,34 @@ 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)

integration/test_collection_batch.py

@@ -1,10 +1,10 @@
 import uuid
 from dataclasses import dataclass
-from typing import Any, Generator, Optional, Protocol, Union
+from typing import Any, Awaitable, Generator, Optional, Protocol, Union
 
 import pytest
 
-from integration.conftest import CollectionFactory, CollectionFactoryGet
+from integration.conftest import AsyncCollectionFactory, CollectionFactory, CollectionFactoryGet
 from weaviate.collections import Collection
 from weaviate.collections.classes.config import (
     Configure,
@@ -17,6 +17,8 @@
 from weaviate.collections.classes.tenants import Tenant
 from weaviate.types import VECTORS
 
+from weaviate.collections.collection.async_ import CollectionAsync
+
 UUID = Union[str, uuid.UUID]
 
 
@@ -55,11 +57,21 @@ def __call__(self, name: str = "", multi_tenancy: bool = False) -> Collection[An
         ...
 
 
+class BatchCollectionAsync(Protocol):
+    """Typing for fixture."""
+
+    def __call__(
+        self, name: str = "", multi_tenancy: bool = False
+    ) -> Awaitable[CollectionAsync[Any, Any]]:
+        """Typing for fixture."""
+        ...
+
+
 @pytest.fixture
 def batch_collection(
     collection_factory: CollectionFactory,
 ) -> Generator[BatchCollection, None, None]:
-    def _factory(name: str = "", multi_tenancy: bool = False) -> Collection[Any, Any]:
+    def _factory(name: str = "", multi_tenancy: bool = False):
         collection = collection_factory(
             name=name,
             vectorizer_config=Configure.Vectorizer.none(),
@@ -78,6 +90,29 @@ def _factory(name: str = "", multi_tenancy: bool = False) -> Collection[Any, Any
     yield _factory
 
 
+@pytest.fixture
+def batch_collection_async(
+    async_collection_factory: AsyncCollectionFactory,
+) -> Generator[BatchCollectionAsync, None, None]:
+    async def _factory(name: str = "", multi_tenancy: bool = False):
+        collection = await async_collection_factory(
+            name=name,
+            vectorizer_config=Configure.Vectorizer.none(),
+            properties=[
+                Property(name="name", data_type=DataType.TEXT),
+                Property(name="age", data_type=DataType.INT),
+            ],
+            multi_tenancy_config=Configure.multi_tenancy(multi_tenancy),
+        )
+        await collection.config.add_reference(
+            ReferenceProperty(name="test", target_collection=collection.name)
+        )
+
+        return collection
+
+    yield _factory
+
+
 @pytest.mark.parametrize(
     "vector",
     [None, [1, 2, 3], MockNumpyTorch([1, 2, 3]), MockTensorFlow([1, 2, 3])],
@@ -233,3 +268,73 @@ def test_non_existant_collection(collection_factory_get: CollectionFactoryGet) -
 
     # above should not throw - depending on the autoschema config this might create an error or
     # not, so we do not check for errors here
+
+
+@pytest.mark.asyncio
+async def test_batch_one_hundred_thousand_objects_async_collection(
+    batch_collection_async: BatchCollectionAsync,
+) -> None:
+    """Test adding one hundred thousand data objects."""
+    col = await batch_collection_async()
+    if col._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 col.batch.stream() as batch:
+        for i in range(nr_objects):
+            await batch.add_object(
+                properties={"name": "test" + str(i)},
+            )
+    end = time.time()
+    print(f"Time taken to add {nr_objects} objects: {end - start} seconds")
+    assert len(col.batch.results.objs.errors) == 0
+    assert len(col.batch.results.objs.all_responses) == nr_objects
+    assert len(col.batch.results.objs.uuids) == nr_objects
+    assert await col.length() == nr_objects
+    assert col.batch.results.objs.has_errors is False
+    assert len(col.batch.failed_objects) == 0, [obj.message for obj in col.batch.failed_objects]
+
+
+@pytest.mark.asyncio
+async def test_ingest_one_hundred_thousand_data_objects_async(
+    batch_collection_async: BatchCollectionAsync,
+) -> None:
+    col = await batch_collection_async()
+    if col._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()
+    results = await col.data.ingest({"name": "test" + str(i)} for i in range(nr_objects))
+    end = time.time()
+    print(f"Time taken to add {nr_objects} objects: {end - start} seconds")
+    assert len(results.errors) == 0
+    assert len(results.all_responses) == nr_objects
+    assert len(results.uuids) == nr_objects
+    assert await col.length() == nr_objects
+    assert results.has_errors is False
+    assert len(results.errors) == 0, [obj.message for obj in results.errors.values()]
+
+
+def test_ingest_one_hundred_thousand_data_objects(
+    batch_collection: BatchCollection,
+) -> None:
+    col = batch_collection()
+    if col._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()
+    results = col.data.ingest({"name": "test" + str(i)} for i in range(nr_objects))
+    end = time.time()
+    print(f"Time taken to add {nr_objects} objects: {end - start} seconds")
+    assert len(results.errors) == 0
+    assert len(results.all_responses) == nr_objects
+    assert len(results.uuids) == nr_objects
+    assert len(col) == nr_objects
+    assert results.has_errors is False
+    assert len(results.errors) == 0, [obj.message for obj in results.errors.values()]

integration/test_rbac.py

@@ -742,10 +742,10 @@ def test_server_side_batching_with_auth() -> None:
     with connect_to_local(
         port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=RBAC_AUTH_CREDS
     ) as client:
-        if client._connection._weaviate_version.is_lower_than(1, 34, 0):
-            pytest.skip("Server-side batching not supported in Weaviate < 1.34.0")
+        if client._connection._weaviate_version.is_lower_than(1, 36, 0):
+            pytest.skip("Server-side batching not supported in Weaviate < 1.36.0")
         collection = client.collections.create(collection_name)
-        with client.batch.experimental() as batch:
+        with client.batch.stream() as batch:
             batch.add_object(collection_name)
             batch.add_object(collection_name)
             batch.add_object(collection_name)