You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ephraimbuddy (via GitHub)" <gi...@apache.org> on 2023/11/28 14:01:10 UTC
Re: [PR] Add helper function for CRUD operations on weaviate's schema and class objects [airflow]
ephraimbuddy commented on code in PR #35919:
URL: https://github.com/apache/airflow/pull/35919#discussion_r1407797436
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -94,22 +99,117 @@ def test_connection(self) -> tuple[bool, str]:
self.log.error("Error testing Weaviate connection: %s", e)
return False, str(e)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
def create_class(self, class_json: dict[str, Any]) -> None:
"""Create a new class."""
client = self.get_client()
client.schema.create_class(class_json)
- def create_schema(self, schema_json: dict[str, Any]) -> None:
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def create_schema(self, schema_json: dict[str, Any] | str) -> None:
"""
Create a new Schema.
Instead of adding classes one by one , you can upload a full schema in JSON format at once.
- :param schema_json: The schema to create
+ :param schema_json: The schema to create or path to the json file holding the schema
"""
client = self.get_client()
client.schema.create(schema_json)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def get_schema(self, class_name: str | None = None):
+ """Get the schema from Weaviate.
+
+ get schema of a class or all classes.
+ """
+ client = self.get_client()
+ return client.schema.get(class_name)
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def delete_class(self, class_name: str):
+ """Delete a schema class from Weaviate. This deletes all associated data."""
+ client = self.get_client()
+ client.schema.delete_class(class_name)
+
+ def delete_schema(
+ self, class_names: list[str] | str | None = None
+ ) -> list[weaviate.UnexpectedStatusCodeException] | None:
+ """
+ Deletes all or specific class if class_names are provided.
+
+ If no class_name is given remove the entire schema from the Weaviate instance and all data associated
+ with it. If class_names are given, delete schema classes from Weaviate. This deletes all associated data.
+ :return: list of error object if any
+ """
+ client = self.get_client()
+ class_names = [class_names] if class_names and isinstance(class_names, str) else class_names
+ if class_names:
+ error_list = []
+ for class_name in class_names:
+ try:
+ self.delete_class(class_name=class_name)
+ except weaviate.UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
Review Comment:
Isn't it better to raise exception instead of returning exception
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -94,22 +99,117 @@ def test_connection(self) -> tuple[bool, str]:
self.log.error("Error testing Weaviate connection: %s", e)
return False, str(e)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
def create_class(self, class_json: dict[str, Any]) -> None:
"""Create a new class."""
client = self.get_client()
client.schema.create_class(class_json)
- def create_schema(self, schema_json: dict[str, Any]) -> None:
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def create_schema(self, schema_json: dict[str, Any] | str) -> None:
"""
Create a new Schema.
Instead of adding classes one by one , you can upload a full schema in JSON format at once.
- :param schema_json: The schema to create
+ :param schema_json: The schema to create or path to the json file holding the schema
"""
client = self.get_client()
client.schema.create(schema_json)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def get_schema(self, class_name: str | None = None):
+ """Get the schema from Weaviate.
+
+ get schema of a class or all classes.
+ """
+ client = self.get_client()
+ return client.schema.get(class_name)
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def delete_class(self, class_name: str):
+ """Delete a schema class from Weaviate. This deletes all associated data."""
+ client = self.get_client()
+ client.schema.delete_class(class_name)
+
+ def delete_schema(
+ self, class_names: list[str] | str | None = None
+ ) -> list[weaviate.UnexpectedStatusCodeException] | None:
+ """
+ Deletes all or specific class if class_names are provided.
+
+ If no class_name is given remove the entire schema from the Weaviate instance and all data associated
+ with it. If class_names are given, delete schema classes from Weaviate. This deletes all associated data.
+ :return: list of error object if any
+ """
+ client = self.get_client()
+ class_names = [class_names] if class_names and isinstance(class_names, str) else class_names
+ if class_names:
+ error_list = []
+ for class_name in class_names:
+ try:
+ self.delete_class(class_name=class_name)
+ except weaviate.UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
+ else:
+ return client.schema.delete_all()
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def update_class(self, class_name: str, config: dict):
+ """Update schema's class."""
+ client = self.get_client()
+ client.schema.update_config(class_name=class_name, config=config)
+
+ def update_multiple_classes(
+ self, schema_json: list[dict]
+ ) -> list[weaviate.UnexpectedStatusCodeException] | None:
+ """Updated multiple classes.
+
+ :param schema_json: list of class_config objects
+ .. seealso:: `example of class_config <https://weaviate-python-client.readthedocs.io/en/v3.25.2/weaviate.schema.html#weaviate.schema.Schema.update_config>`_.
+ :return: list of error object if any
+ """
+ error_list = []
+ for config in schema_json:
+ try:
+ self.update_class(class_name=config.pop("class"), config=config)
+ except weaviate.UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
+
+ def upsert_classes(self, schema_json: dict[str, Any] | str, existing: ExitingSchemaOptions = "ignore"):
+ """
Review Comment:
I personally don't like that we are optimizing hooks for taskflow use
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -94,22 +99,117 @@ def test_connection(self) -> tuple[bool, str]:
self.log.error("Error testing Weaviate connection: %s", e)
return False, str(e)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
def create_class(self, class_json: dict[str, Any]) -> None:
"""Create a new class."""
client = self.get_client()
client.schema.create_class(class_json)
- def create_schema(self, schema_json: dict[str, Any]) -> None:
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def create_schema(self, schema_json: dict[str, Any] | str) -> None:
"""
Create a new Schema.
Instead of adding classes one by one , you can upload a full schema in JSON format at once.
- :param schema_json: The schema to create
+ :param schema_json: The schema to create or path to the json file holding the schema
"""
client = self.get_client()
client.schema.create(schema_json)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def get_schema(self, class_name: str | None = None):
+ """Get the schema from Weaviate.
+
+ get schema of a class or all classes.
+ """
+ client = self.get_client()
+ return client.schema.get(class_name)
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def delete_class(self, class_name: str):
+ """Delete a schema class from Weaviate. This deletes all associated data."""
+ client = self.get_client()
+ client.schema.delete_class(class_name)
+
+ def delete_schema(
+ self, class_names: list[str] | str | None = None
+ ) -> list[weaviate.UnexpectedStatusCodeException] | None:
+ """
+ Deletes all or specific class if class_names are provided.
+
+ If no class_name is given remove the entire schema from the Weaviate instance and all data associated
+ with it. If class_names are given, delete schema classes from Weaviate. This deletes all associated data.
+ :return: list of error object if any
+ """
+ client = self.get_client()
+ class_names = [class_names] if class_names and isinstance(class_names, str) else class_names
+ if class_names:
+ error_list = []
+ for class_name in class_names:
+ try:
+ self.delete_class(class_name=class_name)
+ except weaviate.UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
+ else:
+ return client.schema.delete_all()
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def update_class(self, class_name: str, config: dict):
Review Comment:
This should be called update_config to avoid confusion with update_class_tenants, update_class_shard methods of schema.
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -94,22 +99,117 @@ def test_connection(self) -> tuple[bool, str]:
self.log.error("Error testing Weaviate connection: %s", e)
return False, str(e)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
def create_class(self, class_json: dict[str, Any]) -> None:
"""Create a new class."""
client = self.get_client()
client.schema.create_class(class_json)
- def create_schema(self, schema_json: dict[str, Any]) -> None:
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def create_schema(self, schema_json: dict[str, Any] | str) -> None:
"""
Create a new Schema.
Instead of adding classes one by one , you can upload a full schema in JSON format at once.
- :param schema_json: The schema to create
+ :param schema_json: The schema to create or path to the json file holding the schema
"""
client = self.get_client()
client.schema.create(schema_json)
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def get_schema(self, class_name: str | None = None):
+ """Get the schema from Weaviate.
+
+ get schema of a class or all classes.
+ """
+ client = self.get_client()
+ return client.schema.get(class_name)
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def delete_class(self, class_name: str):
+ """Delete a schema class from Weaviate. This deletes all associated data."""
+ client = self.get_client()
+ client.schema.delete_class(class_name)
+
+ def delete_schema(
+ self, class_names: list[str] | str | None = None
+ ) -> list[weaviate.UnexpectedStatusCodeException] | None:
+ """
+ Deletes all or specific class if class_names are provided.
+
+ If no class_name is given remove the entire schema from the Weaviate instance and all data associated
+ with it. If class_names are given, delete schema classes from Weaviate. This deletes all associated data.
+ :return: list of error object if any
+ """
+ client = self.get_client()
+ class_names = [class_names] if class_names and isinstance(class_names, str) else class_names
+ if class_names:
+ error_list = []
+ for class_name in class_names:
+ try:
+ self.delete_class(class_name=class_name)
+ except weaviate.UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
+ else:
+ return client.schema.delete_all()
+
+ @retry(reraise=True, stop=stop_after_attempt(3), retry=retry_if_exception_type(requests.ConnectionError))
+ def update_class(self, class_name: str, config: dict):
+ """Update schema's class."""
+ client = self.get_client()
+ client.schema.update_config(class_name=class_name, config=config)
+
+ def update_multiple_classes(
+ self, schema_json: list[dict]
+ ) -> list[weaviate.UnexpectedStatusCodeException] | None:
+ """Updated multiple classes.
+
+ :param schema_json: list of class_config objects
+ .. seealso:: `example of class_config <https://weaviate-python-client.readthedocs.io/en/v3.25.2/weaviate.schema.html#weaviate.schema.Schema.update_config>`_.
+ :return: list of error object if any
+ """
+ error_list = []
+ for config in schema_json:
+ try:
+ self.update_class(class_name=config.pop("class"), config=config)
+ except weaviate.UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
Review Comment:
Why are we returning exceptions? Exceptions should be raised or ignored with some logging in my opinion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org