Refactor to separate sync and async connections (#1594)

* Commit working changes (bypassed pre-commit)

* Push non-functional progress (will fail CI)

* Add non-functional backup refactoring

* Add non-functional tenants refactoring

* Add non-functional config refactoring

* Achieve working functionality for everything besides batch

* Merge dev/1.30

* Fix import

* Comment out redundant test

* Fix return types in query executor

* Fix failing backups tests

* Refactor batch to use `ConnectionSync` and `ThreadPoolExecutor`

* Remove redundant event loop code

* Fix event loop import in syncify, rename convert_new to convert

* Fix backups executor, ensure backup tests run serially

* Fix typo

* Fix deprecated `get_assigned_roles`

* Fix auth refactoring to pass the tests

* Fix sync grpc exception handling to pass mock tests

* Fix capitalisation of delete for 1.24

* Reduce boilerplate necessary in `ConnectionX` classes

* Fix missing final raise

* Fix docstring inheritance for client.debug as example

* Introduce new decorator to generate methods from base abstract class

* Migrate users namespace to use impl.generate

* Migrate rbac namespace to use impl.generate

* Refactor collection.data to use impl.generate

* Migrate collection.tenants to impl.generate

* Migration collection.{backups,cluster,config} to impl.generate

* Complete client.backups impl.generate migration

* Migrate collection.{generate,query} to use impl.generate

* Migrate collection.aggregate to impl.generate pattern

* Remove syncify code

* Update pyright in CI and remove mypy

* Fix outstanding bugs and migration client to impl.generate

* Fix imports of backup

* Fix more backup import issues

* Fix typo in new dev/1.30 code

* Make backup easier

* Second try

* Fix typing errprs

* Remove composition of executor completely

* Migrate rest of codebase to inheriting from `_Executor` and new `impl.generate` functinoality

* Align executor and interface surfaces

* Fix backup inheritance and rename/move wrapping decorators

* Fix default args in users executor

* Remove `_XBase` classes by making executors generic on `ConnectionType`

* Simplify semantics of importing from `connect.executor`

* Add QA ignores for flake8 in CI

* Fix inheritance in interface

* Respond to review

* Tidy up import paths and remove `_WeaviateClientBase`

* Fix default arg for `revoke_key` in `deactivate`

* Add journey test for gevent+wsgi usage, add back litestar test

* Fix breaking public imports from .users/.backups reducing diff

* Have `delete_many` use specific grpc class rather than custom impl

* Decrease diff in tests

* Add changelog for refactor

---------

Co-authored-by: Dirk Kulawiak <dirk@semi.technology>

Author: tsmith023

.github/workflows/main.yaml

@@ -65,11 +65,9 @@ jobs:
           python-version: ${{ matrix.version }}
           cache: 'pip' # caching pip dependencies
       - run: pip install -r requirements-devel.txt
-      - name: Run mypy
-        run: mypy --config-file ./pyproject.toml --warn-unused-ignores --python-version ${{matrix.version}} ${{ matrix.folder }}
       - uses: jakebailey/pyright-action@v2
         with:
-          version: 1.1.347
+          version: 1.1.398
 
   unit-tests:
     name: Run Unit Tests
@@ -169,7 +167,7 @@ jobs:
         run: pytest -n auto --dist loadgroup -v --cov --cov-report=term-missing --cov=weaviate --cov-report xml:coverage-integration.xml integration
       - name: Run integration tests without auth secrets (for forks)
         if: ${{ github.event.pull_request.head.repo.fork }}
-        run: pytest -v --cov --cov-report=term-missing --cov=weaviate --cov-report xml:coverage-integration.xml integration
+        run: pytest -n auto --dist loadgroup -v --cov --cov-report=term-missing --cov=weaviate --cov-report xml:coverage-integration.xml integration
       - name: Archive code coverage results
         if: matrix.versions.py == '3.10' && (github.ref_name != 'main')
         uses: actions/upload-artifact@v4

.pre-commit-config.yaml

@@ -36,19 +36,6 @@ repos:
       ]
       files: '^weaviate/collections'
 
-- repo: local
-  hooks:
-    - id: mypy
-      name: mypy
-      entry: ./run-mypy.sh
-      language: python
-      language_version: "3.12"
-      # use require_serial so that script
-      # is only called once per commit
-      require_serial: true
-      # Print the number of files as a sanity-check
-      verbose: true
-
 - repo: local
   hooks:
     - id: pyright
@@ -57,4 +44,4 @@ repos:
       language: node
       pass_filenames: false
       types: [python]
-      additional_dependencies: [pyright@1.1.364]
+      additional_dependencies: [pyright@1.1.398]

docs/changelog.rst

@@ -1,6 +1,14 @@
 Changelog
 =========
 
+Version 4.13.0
+--------------
+This minor version provides a complete top-down refactor of the underlying transport for the sync and async clients.
+    - Introduction of a dependency injection system whereby the sync/async connections are injected into each client independently thereby removing their cross-dependency.
+    - Removal of the ``event_loop.py`` file and all logic surrounding its usage by the sync client.
+    - Refactoring of the batching algorithm to use blocking threads with a sync connection, rather than the event loop sidecar thread.
+
+
 Version 4.12.1
 --------------
 This patch version includes:

integration/test_auth.py

@@ -137,6 +137,9 @@ def test_client_with_authentication_with_anon_weaviate() -> None:
     with pytest.warns(UserWarning) as recwarn:
         with weaviate.connect_to_local(auth_credentials=auth) as client:
             client.collections.list_all()
+        if len(recwarn) > 1:
+            for rwarning in recwarn.list:
+                print(rwarning.message)
         assert len(recwarn) == 1
         assert str(recwarn.list[0].message).startswith("Auth001")
 

integration/test_backup_v4.py

@@ -356,7 +356,7 @@ def test_backup_and_restore_with_collection(
 
     conf_create: Optional[wvc.backup.BackupConfigCreate] = None
     conf_restore: Optional[wvc.backup.BackupConfigRestore] = None
-    backup_location: Optional[wvc.backup.BackupLocation] = None
+    backup_location: Optional[wvc.backup.BackupLocationType] = None
     if dynamic_backup_location:
         if client._connection._weaviate_version.is_lower_than(1, 27, 2):
             pytest.skip("Cancel backups is only supported from 1.27.2")
@@ -505,7 +505,7 @@ def test_cancel_backup(
     if client._connection._weaviate_version.is_lower_than(1, 24, 25):
         pytest.skip("Cancel backups is only supported from 1.24.25")
 
-    backup_location: Optional[wvc.backup.BackupLocation] = None
+    backup_location: Optional[wvc.backup.BackupLocationType] = None
     if dynamic_backup_location:
         if client._connection._weaviate_version.is_lower_than(1, 27, 2):
             pytest.skip("Cancel backups is only supported from 1.27.2")

integration/test_client_debug.py

@@ -1,4 +1,10 @@
-from integration.conftest import ClientFactory, CollectionFactory
+import pytest
+from integration.conftest import (
+    AsyncClientFactory,
+    AsyncCollectionFactory,
+    ClientFactory,
+    CollectionFactory,
+)
 
 from weaviate.classes.config import DataType, Property
 from weaviate.classes.debug import DebugRESTObject
@@ -22,6 +28,27 @@ def test_get_object_single_node(
     assert debug_obj is None
 
 
+@pytest.mark.asyncio
+async def test_get_object_single_node_async(
+    async_client_factory: AsyncClientFactory, async_collection_factory: AsyncCollectionFactory
+) -> None:
+    client = await async_client_factory()
+    collection = await async_collection_factory(
+        properties=[Property(name="name", data_type=DataType.TEXT)]
+    )
+
+    uuid = await collection.data.insert({"name": "John Doe"})
+
+    debug_obj = await client.debug.get_object_over_rest(collection.name, uuid)
+    assert debug_obj is not None
+    assert isinstance(debug_obj, DebugRESTObject)
+    assert str(debug_obj.uuid) == str(uuid)
+
+    non_existant_uuid = "00000000-0000-0000-0000-000000000000"
+    debug_obj = await client.debug.get_object_over_rest(collection.name, non_existant_uuid)
+    assert debug_obj is None
+
+
 def test_get_object_multi_node(
     client_factory: ClientFactory, collection_factory: CollectionFactory
 ) -> None:

integration/test_collection_batch_delete.py

@@ -19,6 +19,7 @@
 )
 from weaviate.collections.classes.internal import ReferenceToMulti
 from weaviate.collections.classes.tenants import Tenant
+from weaviate.classes.config import ConsistencyLevel
 from weaviate.exceptions import (
     WeaviateQueryException,
 )
@@ -633,3 +634,26 @@ def test_delete_by_time_metadata_with_ref(
     assert len(source) == 1
     assert source.query.fetch_object_by_id(uuid=uuid_source1) is not None
     assert source.query.fetch_object_by_id(uuid=uuid_source2) is None
+
+
+def test_delete_many_with_consistency_level(
+    collection_factory: CollectionFactory,
+) -> None:
+    collection = collection_factory(
+        ports=(8087, 50058),
+        properties=[
+            Property(name="text", data_type=DataType.TEXT),
+            Property(name="int", data_type=DataType.INT),
+        ],
+    ).with_consistency_level(ConsistencyLevel.ALL)
+    collection.data.insert_many(
+        [
+            DataObject(properties={"int": 10}, uuid=UUID1),
+            DataObject(properties={"text": "I am ageless"}, uuid=UUID2),
+        ]
+    )
+    assert len(collection.query.fetch_objects().objects) == 2
+
+    collection.data.delete_many(where=Filter.by_id().equal(UUID1))
+    objects = collection.query.fetch_objects().objects
+    assert len(objects) == 1

integration/test_rbac.py

@@ -16,11 +16,10 @@
     NodesPermissionOutput,
     TenantsPermissionOutput,
     UsersPermissionOutput,
+    UserTypes,
 )
 from _pytest.fixtures import SubRequest
 
-from weaviate.users.users import UserTypes
-
 RBAC_PORTS = (8092, 50063)
 RBAC_AUTH_CREDS = Auth.api_key("admin-key")
 

journey_tests/gunicorn/app.py journey_tests/gunicorn/asgi.pyjourney_tests/gunicorn/app.py renamed to journey_tests/gunicorn/asgi.py

@@ -0,0 +1,22 @@
+from typing import List
+
+from flask import Flask, g
+
+from journey_tests.journeys import SyncJourneys
+
+app = Flask(__name__)
+
+
+def get_sync_client() -> SyncJourneys:
+    if "sync" not in g:
+        g.sync = SyncJourneys.use()
+    return g.sync
+
+
+@app.route("/sync-in-sync")
+def sync() -> List[dict]:
+    return get_sync_client().simple()
+
+
+if __name__ == "__main__":
+    app.run(debug=False, host="0.0.0.0")