pankajkoti commented on code in PR #35919:
URL: https://github.com/apache/airflow/pull/35919#discussion_r1410245744
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -112,22 +117,170 @@ 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[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 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[UnexpectedStatusCodeException] | None:
Review Comment:
This method looks redundant. Either it could be handled directly in
update_class above my modifying the signature of params to accept a single
class or multiple class JSONs or the user could use a loop to call update_class
on the list of classes.
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -112,22 +117,170 @@ 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.
Review Comment:
what does all classes mean here?
##########
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:
+1, why to return a list of errors? how will these errors be handled later?
I think we should accept an argument here which tells what to do. e.g in the
case when multiple classes are given for deletion, based on the argument value,
while iterating it should either continue or raise exception when deletion of a
class fails and maybe return a json saying how many failed or which ones failed
deletion, how many successfully deleted.
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -112,22 +117,170 @@ 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(
Review Comment:
This method name and the params it accepts look off. When the method name is
delete schema, it should accept schema name.
For deleting a class or list of classes, there could be a separate method
delete_class accepting class_name(s)
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -112,22 +117,170 @@ 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[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 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[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 UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
+
+ def upsert_classes(self, schema_json: dict[str, Any] | str, existing:
ExitingSchemaOptions = "ignore"):
+ """
+ Create or update the classes in schema of Weaviate database.
+
+ :param schema_json: Json containing the schema. Format {"class_name":
"class_dict"}
+ .. seealso:: `example of class_dict
<https://weaviate-python-client.readthedocs.io/en/v3.25.2/weaviate.schema.html#weaviate.schema.Schema.create>`_.
+ :param existing: Options to handle the case when the classes exist,
possible options
+ 'replace', 'fail', 'ignore'.
+ """
+ existing_schema_options = ["replace", "fail", "ignore"]
+ if existing not in existing_schema_options:
+ raise ValueError(f"Param 'existing' should be one of the
{existing_schema_options} values.")
+ if isinstance(schema_json, str):
+ schema_json = cast(dict, json.load(open(schema_json)))
+ set__exiting_classes = {class_object["class"] for class_object in
self.get_schema()["classes"]}
+ set__to_be_added_classes = {key for key, _ in schema_json.items()}
+ intersection_classes =
set__exiting_classes.intersection(set__to_be_added_classes)
+ classes_to_create = set()
+ if existing == "fail" and intersection_classes:
+ raise ValueError(
+ f"Trying to create class {intersection_classes}" f" but this
class already exists."
+ )
+ elif existing == "ignore":
+ classes_to_create = set__to_be_added_classes - set__exiting_classes
+ elif existing == "replace":
+ error_list =
self.delete_schema(class_names=list(intersection_classes))
+ if error_list:
+ raise ValueError(error_list)
+ classes_to_create =
intersection_classes.union(set__to_be_added_classes)
+ classes_to_create_list = [schema_json[item] for item in
sorted(list(classes_to_create))]
+ self.create_schema({"classes": classes_to_create_list})
+
+ def _compare_schema_subset(self, class_object: Any, class_schema: Any) ->
bool:
Review Comment:
The param names seem to be confusing here. In some method above we called
config as a param. Is the config same as object here? Can we have consistent
param names.
What is class_schema? Can it just be called schema. The docstring says this
is current schema. Then can current schema be not retrieved using self?
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -112,22 +117,170 @@ 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[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 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[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 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:
There seems to be a lot of redundancy for creation of classes. We have
create_schema, create_class, update_class, update_multiple_classes,
upsert_classes all trying to manipulate classes
We should try to consolidate them instead of building separate methods.
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -112,22 +117,170 @@ 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[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 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[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 UnexpectedStatusCodeException as e:
+ error_list.append(e)
+ return error_list
+
+ def upsert_classes(self, schema_json: dict[str, Any] | str, existing:
ExitingSchemaOptions = "ignore"):
+ """
+ Create or update the classes in schema of Weaviate database.
+
+ :param schema_json: Json containing the schema. Format {"class_name":
"class_dict"}
+ .. seealso:: `example of class_dict
<https://weaviate-python-client.readthedocs.io/en/v3.25.2/weaviate.schema.html#weaviate.schema.Schema.create>`_.
+ :param existing: Options to handle the case when the classes exist,
possible options
+ 'replace', 'fail', 'ignore'.
+ """
+ existing_schema_options = ["replace", "fail", "ignore"]
+ if existing not in existing_schema_options:
+ raise ValueError(f"Param 'existing' should be one of the
{existing_schema_options} values.")
+ if isinstance(schema_json, str):
+ schema_json = cast(dict, json.load(open(schema_json)))
+ set__exiting_classes = {class_object["class"] for class_object in
self.get_schema()["classes"]}
+ set__to_be_added_classes = {key for key, _ in schema_json.items()}
+ intersection_classes =
set__exiting_classes.intersection(set__to_be_added_classes)
+ classes_to_create = set()
+ if existing == "fail" and intersection_classes:
+ raise ValueError(
+ f"Trying to create class {intersection_classes}" f" but this
class already exists."
+ )
+ elif existing == "ignore":
+ classes_to_create = set__to_be_added_classes - set__exiting_classes
+ elif existing == "replace":
+ error_list =
self.delete_schema(class_names=list(intersection_classes))
+ if error_list:
+ raise ValueError(error_list)
+ classes_to_create =
intersection_classes.union(set__to_be_added_classes)
+ classes_to_create_list = [schema_json[item] for item in
sorted(list(classes_to_create))]
+ self.create_schema({"classes": classes_to_create_list})
+
+ def _compare_schema_subset(self, class_object: Any, class_schema: Any) ->
bool:
+ """
+ Recursively check if requested schema/object is a subset of the
current schema.
+
+ :param class_object: The class object to check against current schema
+ :param class_schema: The current schema class object
+ """
+ # Direct equality check
+ if class_object == class_schema:
+ return True
+
+ # Type mismatch early return
+ if type(class_object) != type(class_schema):
+ return False
+
+ # Dictionary comparison
+ if isinstance(class_object, dict):
+ for k, v in class_object.items():
+ if (k not in class_schema) or (not
self._compare_schema_subset(v, class_schema[k])):
+ return False
+ return True
+
+ # List or Tuple comparison
+ if isinstance(class_object, (list, tuple)):
+ for obj, sch in zip(class_object, class_schema):
+ if len(class_object) > len(class_schema) or not
self._compare_schema_subset(obj, sch):
+ return False
+ return True
+
+ # Default case for non-matching types or unsupported types
+ return False
+
+ @staticmethod
+ def _convert_properties_to_dict(classes_objects, key_property: str =
"name"):
Review Comment:
can we add dosctring for this method. what is the significance of this
method? what are properties here?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
