Upgrade to weaviate-client to v4

airflow/providers/weaviate/CHANGELOG.rst

@@ -20,8 +20,24 @@
 Changelog
 ---------
 
-1.4.2
-.....
+2.0.0
+......
+
+
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+.. warning::
+   * We bumped the minimum version of weaviate-client to 4.4.0. Many of the concepts and methods have been changed. We suggest you read `Migrate from v3 to v4 <https://weaviate.io/developers/weaviate/client-libraries/python/v3_v4_migration>`_ before you upgrade to this version
+
+* Add columns ``Port``, ``gRPC host``, ``gRPC port``  and ``Use https``, ``Use a secure channel for the underlying gRPC API`` options  to the Weaviate connection. The default values from Airflow providers may not be suitable for using Weaviate correctly, so we recommend explicitly specifying these values.
+* Update ``WeaviateIngestOperator`` and ``WeaviateDocumentIngestOperator`` to use ``WeaviateHook`` with ``weaviate-client`` v4 API. The major changes are changing argument ``class_name`` to ``collection_name`` and removing ``batch_params``.
+* Update ``WeaviateHook`` to utilize ``weaviate-client`` v4 API. The implementation has been extensively changed. We recommend reading `Migrate from v3 to v4 <https://weaviate.io/developers/weaviate/client-libraries/python/v3_v4_migration>`_ to understand the changes on the Weaviate side before using the updated ``WeaviateHook``.
+* Migrate the following ``WeaviateHook`` public methods to v4 API: ``test_connections``, ``query_with_vector``, ``create_object``, ``get_object``, ``delete_object``, ``update_object``, ``replace_object``, ``object_exists``, ``batch_data``, ``get_or_create_object``, ``create_or_replace_document_objects``
+* Rename ``WeaviateHook`` public methods ``update_schema`` as ``update_collection_configuration``, ``create_class`` as ``create_collection``, ``get_schema`` as ``get_collection_configuraiton``, ``delete_classes`` as ``delete_collections`` and ``query_without_vector`` as ``query_with_text``.
+* Remove the following ``WeaviateHook`` public methods: ``validate_object``, ``update_schema``, ``create_schema``, ``delete_all_schema``, ``check_subset_of_schema``
+* Remove deprecated method ``WeaviateHook.get_client``
+* Remove unused argument ``retry_status_codes`` in ``WeaviateHook.__init__``
 
 Misc
 ~~~~

airflow/providers/weaviate/operators/weaviate.py

@@ -27,6 +27,7 @@
 
 if TYPE_CHECKING:
     import pandas as pd
+    from weaviate.types import UUID
 
     from airflow.utils.context import Context
 
@@ -43,11 +44,10 @@ class WeaviateIngestOperator(BaseOperator):
     custom vectors and store them in the Weaviate class.
 
     :param conn_id: The Weaviate connection.
-    :param class_name: The Weaviate class to be used for storing the data objects into.
+    :param collection: The Weaviate collection to be used for storing the data objects into.
     :param input_data: The list of dicts or pandas dataframe representing Weaviate data objects to generate
         embeddings on (or provides custom vectors) and store them in the Weaviate class.
     :param vector_col: key/column name in which the vectors are stored.
-    :param batch_params: Additional parameters for Weaviate batch configuration.
     :param hook_params: Optional config params to be passed to the underlying hook.
         Should match the desired hook constructor params.
     :param input_json: (Deprecated) The JSON representing Weaviate data objects to generate embeddings on
@@ -59,25 +59,23 @@ class WeaviateIngestOperator(BaseOperator):
     def __init__(
         self,
         conn_id: str,
-        class_name: str,
+        collection_name: str,
         input_data: list[dict[str, Any]] | pd.DataFrame | None = None,
         vector_col: str = "Vector",
         uuid_column: str = "id",
         tenant: str | None = None,
-        batch_params: dict | None = None,
         hook_params: dict | None = None,
         input_json: list[dict[str, Any]] | pd.DataFrame | None = None,
         **kwargs: Any,
     ) -> None:
         super().__init__(**kwargs)
-        self.class_name = class_name
+        self.collection_name = collection_name
         self.conn_id = conn_id
         self.vector_col = vector_col
         self.input_json = input_json
         self.uuid_column = uuid_column
         self.tenant = tenant
         self.input_data = input_data
-        self.batch_params = batch_params or {}
         self.hook_params = hook_params or {}
 
         if (self.input_data is None) and (input_json is not None):
@@ -96,18 +94,14 @@ def hook(self) -> WeaviateHook:
         """Return an instance of the WeaviateHook."""
         return WeaviateHook(conn_id=self.conn_id, **self.hook_params)
 
-    def execute(self, context: Context) -> list:
+    def execute(self, context: Context) -> None:
         self.log.debug("Input data: %s", self.input_data)
-        insertion_errors: list = []
         self.hook.batch_data(
-            class_name=self.class_name,
+            collection_name=self.collection_name,
             data=self.input_data,
-            batch_config_params=self.batch_params,
             vector_col=self.vector_col,
             uuid_col=self.uuid_column,
-            tenant=self.tenant,
         )
-        return insertion_errors
 
 
 class WeaviateDocumentIngestOperator(BaseOperator):
@@ -132,12 +126,11 @@ class WeaviateDocumentIngestOperator(BaseOperator):
     error: raise an error if an object belonging to a existing document is tried to be created.
 
     :param data: A single pandas DataFrame or a list of dicts to be ingested.
-    :param class_name: Name of the class in Weaviate schema where data is to be ingested.
+    :param collection_name: Name of the collection in Weaviate schema where data is to be ingested.
     :param existing: Strategy for handling existing data: 'skip', or 'replace'. Default is 'skip'.
     :param document_column: Column in DataFrame that identifying source document.
     :param uuid_column: Column with pre-generated UUIDs. If not provided, UUIDs will be generated.
     :param vector_column: Column with embedding vectors for pre-embedded data.
-    :param batch_config_params: Additional parameters for Weaviate batch configuration.
     :param tenant: The tenant to which the object will be added.
     :param verbose: Flag to enable verbose output during the ingestion process.
     :param hook_params: Optional config params to be passed to the underlying hook.
@@ -150,12 +143,11 @@ def __init__(
         self,
         conn_id: str,
         input_data: pd.DataFrame | list[dict[str, Any]] | list[pd.DataFrame],
-        class_name: str,
+        collection_name: str,
         document_column: str,
         existing: str = "skip",
         uuid_column: str = "id",
         vector_col: str = "Vector",
-        batch_config_params: dict | None = None,
         tenant: str | None = None,
         verbose: bool = False,
         hook_params: dict | None = None,
@@ -164,12 +156,11 @@ def __init__(
         super().__init__(**kwargs)
         self.conn_id = conn_id
         self.input_data = input_data
-        self.class_name = class_name
+        self.collection_name = collection_name
         self.document_column = document_column
         self.existing = existing
         self.uuid_column = uuid_column
         self.vector_col = vector_col
-        self.batch_config_params = batch_config_params
         self.tenant = tenant
         self.verbose = verbose
         self.hook_params = hook_params or {}
@@ -179,22 +170,20 @@ def hook(self) -> WeaviateHook:
         """Return an instance of the WeaviateHook."""
         return WeaviateHook(conn_id=self.conn_id, **self.hook_params)
 
-    def execute(self, context: Context) -> list:
+    def execute(self, context: Context) -> Sequence[dict[str, UUID | str] | None]:
         """
         Create or replace objects belonging to documents.
 
         :return: List of UUID which failed to create
         """
         self.log.debug("Total input objects : %s", len(self.input_data))
-        insertion_errors = self.hook.create_or_replace_document_objects(
+        batch_delete_error = self.hook.create_or_replace_document_objects(
             data=self.input_data,
-            class_name=self.class_name,
+            collection_name=self.collection_name,
             document_column=self.document_column,
             existing=self.existing,
             uuid_column=self.uuid_column,
             vector_column=self.vector_col,
-            batch_config_params=self.batch_config_params,
-            tenant=self.tenant,
             verbose=self.verbose,
         )
-        return insertion_errors
+        return batch_delete_error

airflow/providers/weaviate/provider.yaml

@@ -28,7 +28,7 @@ source-date-epoch: 1718605569
 
 # note that those versions are maintained by release manager - do not update them manually
 versions:
-  - 1.4.2
+  - 2.0.0
   - 1.4.1
   - 1.4.0
   - 1.3.4
@@ -50,7 +50,7 @@ integrations:
 dependencies:
   - apache-airflow>=2.7.0
   - httpx>=0.25.0
-  - weaviate-client>=3.24.2
+  - weaviate-client>=4.4.0
   # In pandas 2.2 minimal version of the sqlalchemy is 2.0
   # https://pandas.pydata.org/docs/whatsnew/v2.2.0.html#increased-minimum-versions-for-dependencies
   # However Airflow not fully supports it yet: https://github.com/apache/airflow/issues/28723

docs/apache-airflow-providers-weaviate/connections.rst

@@ -31,14 +31,17 @@ Configuring the Connection
 --------------------------
 
 Host (required)
-    Host URL to connect to the Weaviate cluster.
+    The host to use for the Weaviate cluster REST and GraphQL API calls. DO NOT include the schema (i.e., http or https).
 
 OIDC Username (optional)
     Username for the OIDC user when OIDC option is to be used for authentication.
 
 OIDC Password (optional)
     Password for the OIDC user when OIDC option is to be used for authentication.
 
+Port (option)
+    The port to use for the Weaviate cluster REST and GraphQL API calls.
+
 Extra (optional)
     Specify the extra parameters (as json dictionary) that can be used in the
     connection. All parameters are optional.
@@ -48,11 +51,24 @@ Extra (optional)
     * If you'd like to use Vectorizers for your class, configure the API keys to use the corresponding
       embedding API. The extras accepts a key ``additional_headers`` containing the dictionary
       of API keys for the embedding API authentication. They are mentioned in a section here:
-      `addtional_headers <https://weaviate.io/developers/academy/zero_to_mvp/hello_weaviate/hands_on#-client-instantiation>`__
+      `Third party API keys <https://weaviate.io/developers/weaviate/starter-guides/connect#third-party-api-keys>`__
 
 Weaviate API Token (optional)
     Specify your Weaviate API Key to connect when API Key option is to be used for authentication.
 
+Use https (optional)
+    Whether to use https for the Weaviate cluster REST and GraphQL API calls.
+
+gRPC host (optional)
+    The host to use for the Weaviate cluster gRPC API.
+
+gRPC port (optional)
+    The port to use for the Weaviate cluster gRPC API.
+
+Use a secure channel for the underlying gRPC API (optional)
+    Whether to use a secure channel for the the Weaviate cluster gRPC API.
+
+
 Supported Authentication Methods
 --------------------------------
 * API Key Authentication: This method uses the Weaviate API Key to authenticate the connection. You can either have the

docs/apache-airflow-providers-weaviate/index.rst

@@ -104,7 +104,7 @@ PIP package          Version required
 ===================  =========================================
 ``apache-airflow``   ``>=2.7.0``
 ``httpx``            ``>=0.25.0``
-``weaviate-client``  ``>=3.24.2``
+``weaviate-client``  ``>=4.4.0``
 ``pandas``           ``>=2.1.2,<2.2; python_version >= "3.9"``
 ``pandas``           ``>=1.5.3,<2.2; python_version < "3.9"``
 ===================  =========================================

generated/provider_dependencies.json

@@ -1319,7 +1319,7 @@
       "httpx>=0.25.0",
       "pandas>=1.5.3,<2.2;python_version<\"3.9\"",
       "pandas>=2.1.2,<2.2;python_version>=\"3.9\"",
-      "weaviate-client>=3.24.2"
+      "weaviate-client>=4.4.0"
     ],
     "devel-deps": [],
     "plugins": [],