uranusjr commented on code in PR #36085:
URL: https://github.com/apache/airflow/pull/36085#discussion_r1417028234
##########
airflow/providers/weaviate/hooks/weaviate.py:
##########
@@ -135,22 +139,40 @@ def create_schema(self, schema_json: dict[str, Any]) ->
None:
client = self.conn
client.schema.create(schema_json)
+ @staticmethod
+ def check_http_error_is_retryable(exc: BaseException):
+ return isinstance(exc, requests.HTTPError) and not exc.response.ok
+
def batch_data(
- self, class_name: str, data: list[dict[str, Any]],
batch_config_params: dict[str, Any] | None = None
+ self,
+ class_name: str,
+ data: list[dict[str, Any]] | pd.DataFrame,
+ batch_config_params: dict[str, Any] | None = None,
+ vector_col: str = "Vector",
+ no_retry_attempts_per_object: int = 5,
) -> None:
client = self.conn
if not batch_config_params:
batch_config_params = {}
client.batch.configure(**batch_config_params)
+ if isinstance(data, pd.DataFrame):
+ data = json.loads(data.to_json(orient="records"))
with client.batch as batch:
# Batch import all data
for index, data_obj in enumerate(data):
- self.log.debug("importing data: %s", index + 1)
- vector = data_obj.pop("Vector", None)
- if vector is not None:
- batch.add_data_object(data_obj, class_name, vector=vector)
- else:
- batch.add_data_object(data_obj, class_name)
+ for attempt in Retrying(
+ stop=stop_after_attempt(no_retry_attempts_per_object),
+ retry=retry_if_exception(lambda exc:
self.check_http_error_is_retryable(exc)),
Review Comment:
```suggestion
retry=retry_if_exception(self.check_http_error_is_retryable),
```
--
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-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
