class_docstr: Interact with Weaviate database to store vectors. This hook uses the 'conn_id'.
  
  :param conn_id: The connection id to use when connecting to Weaviate. <howto/connection:weaviate>
class_docstr_tok: ["Interact","with","Weaviate","database","to","store","vectors",".","This","hook","uses","the","'conn_id","'",".",":","param","conn_id",":","The","connection","id","to","use","when","connecting","to","Weaviate",".","<","howto\/connection",":","weaviate",">"]
docstr: create or replace objects belonging to documents.
  
  In real-world scenarios, information sources like Airflow docs, Stack Overflow, or other issues
  are considered 'documents' here. It's crucial to keep the database objects in sync with these sources.
  If any changes occur in these documents, this function aims to reflect those changes in the database.
  
  .. note::
  
      This function assumes responsibility for identifying changes in documents, dropping relevant
      database objects, and recreating them based on updated information. It's crucial to handle this
      process with care, ensuring backups and validation are in place to prevent data loss or
      inconsistencies.
  
  Provides users with multiple ways of dealing with existing values.
  replace: replace the existing objects with new objects. This option requires to identify the
  objects belonging to a document. which by default is done by using document_column field.
  skip: skip the existing objects and only add the missing objects of a document.
  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 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.
  :return: list of UUID which failed to create
docstr_tok: ["create","or","replace","objects","belonging","to","documents",".","In","real-world","scenarios",",","information","sources","like","Airflow","docs",",","Stack","Overflow",",","or","other","issues","are","considered","'documents","'","here",".","It","'s","crucial","to","keep","the","database","objects","in","sync","with","these","sources",".","If","any","changes","occur","in","these","documents",",","this","function","aims","to","reflect","those","changes","in","the","database",".","..","note",":",":","This","function","assumes","responsibility","for","identifying","changes","in","documents",",","dropping","relevant","database","objects",",","and","recreating","them","based","on","updated","information",".","It","'s","crucial","to","handle","this","process","with","care",",","ensuring","backups","and","validation","are","in","place","to","prevent","data","loss","or","inconsistencies",".","Provides","users","with","multiple","ways","of","dealing","with","existing","values",".","replace",":","replace","the","existing","objects","with","new","objects",".","This","option","requires","to","identify","the","objects","belonging","to","a","document",".","which","by","default","is","done","by","using","document_column","field",".","skip",":","skip","the","existing","objects","and","only","add","the","missing","objects","of","a","document",".","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","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",".",":","return",":","list","of","UUID","which","failed","to","create"]
code: def create_or_replace_document_objects(
      self,
      data: pd.DataFrame | list[dict[str, Any]] | list[pd.DataFrame],
      class_name: str,
      document_column: str,
      existing: str = "skip",
      uuid_column: str | None = None,
      vector_column: str = "Vector",
      batch_config_params: dict | None = None,
      tenant: str | None = None,
      verbose: bool = False,
  ):
      """
      create or replace objects belonging to documents.
  
      In real-world scenarios, information sources like Airflow docs, Stack Overflow, or other issues
      are considered 'documents' here. It's crucial to keep the database objects in sync with these sources.
      If any changes occur in these documents, this function aims to reflect those changes in the database.
  
      .. note::
  
          This function assumes responsibility for identifying changes in documents, dropping relevant
          database objects, and recreating them based on updated information. It's crucial to handle this
          process with care, ensuring backups and validation are in place to prevent data loss or
          inconsistencies.
  
      Provides users with multiple ways of dealing with existing values.
      replace: replace the existing objects with new objects. This option requires to identify the
      objects belonging to a document. which by default is done by using document_column field.
      skip: skip the existing objects and only add the missing objects of a document.
      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 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.
      :return: list of UUID which failed to create
      """
      import pandas as pd
  
      if existing not in ["skip", "replace", "error"]:
          raise ValueError(
              "Invalid parameter for 'existing'. Choices are 'skip', 'replace', 'error'."
          )
  
      if len(data) == 0:
          return []
  
      if isinstance(data, Sequence) and isinstance(data[0], dict):
          # This is done to narrow the type to List[Dict[str, Any].
          data = pd.json_normalize(cast(List[Dict[str, Any]], data))
      elif isinstance(data, Sequence) and isinstance(
          data[0], pd.DataFrame
      ):
          # This is done to narrow the type to List[pd.DataFrame].
          data = pd.concat(
              cast(List[pd.DataFrame], data), ignore_index=True
          )
      else:
          data = cast(pd.DataFrame, data)
  
      unique_columns = sorted(data.columns.to_list())
  
      if verbose:
          self.log.info(
              "%s objects came in for insertion.", data.shape[0]
          )
  
      if uuid_column is None or uuid_column not in data.columns:
          (
              data,
              uuid_column,
          ) = self._generate_uuids(
              df=data,
              class_name=class_name,
              unique_columns=unique_columns,
              vector_column=vector_column,
              uuid_column=uuid_column,
          )
  
      # drop duplicate rows, using uuid_column and unique_columns. Removed  `None` as it can be added to
      # set when `uuid_column` is None.
      data = data.drop_duplicates(
          subset=[document_column, uuid_column], keep="first"
      )
      if verbose:
          self.log.info(
              "%s objects remain after deduplication.", data.shape[0]
          )
  
      batch_delete_error: list = []
      (
          documents_to_uuid_map,
          changed_documents,
          unchanged_documents,
          new_documents,
      ) = self._get_segregated_documents(
          data=data,
          document_column=document_column,
          uuid_column=uuid_column,
          class_name=class_name,
      )
      if verbose:
          self.log.info(
              "Found %s changed documents, %s unchanged documents and %s non-existing documents",
              len(changed_documents),
              len(unchanged_documents),
              len(new_documents),
          )
          for document in changed_documents:
              self.log.info(
                  "Changed document: %s has %s objects.",
                  document,
                  len(documents_to_uuid_map[document]),
              )
  
          self.log.info(
              "Non-existing document: %s", ", ".join(new_documents)
          )
  
      if existing == "error" and len(changed_documents):
          raise ValueError(
              f"Documents {', '.join(changed_documents)} already exists. You can either skip or replace"
              f" them by passing 'existing=skip' or 'existing=replace' respectively."
          )
      elif existing == "skip":
          data = data[data[document_column].isin(new_documents)]
          if verbose:
              self.log.info(
                  "Since existing=skip, ingesting only non-existing document's object %s",
                  data.shape[0],
              )
      elif existing == "replace":
          total_objects_count = sum(
              [
                  len(documents_to_uuid_map[doc])
                  for doc in changed_documents
              ]
          )
          if verbose:
              self.log.info(
                  "Since existing='replace', deleting %s objects belonging changed documents %s",
                  total_objects_count,
                  changed_documents,
              )
          batch_delete_error = self._delete_all_documents_objects(
              document_keys=list(changed_documents),
              total_objects_count=total_objects_count,
              document_column=document_column,
              class_name=class_name,
              batch_delete_error=batch_delete_error,
              tenant=tenant,
              batch_config_params=batch_config_params,
              verbose=verbose,
          )
          data = data[
              data[document_column].isin(
                  new_documents.union(changed_documents)
              )
          ]
          self.log.info(
              "Batch inserting %s objects for non-existing and changed documents.",
              data.shape[0],
          )
  
      insertion_errors: list = []
      if data.shape[0]:
          insertion_errors = self.batch_data(
              class_name=class_name,
              data=data,
              batch_config_params=batch_config_params,
              vector_col=vector_column,
              uuid_col=uuid_column,
              tenant=tenant,
          )
          if insertion_errors or batch_delete_error:
              if insertion_errors:
                  self.log.info(
                      "Failed to insert %s objects.",
                      len(insertion_errors),
                  )
              if batch_delete_error:
                  self.log.info(
                      "Failed to delete %s objects.",
                      len(insertion_errors),
                  )
              # Rollback object that were not created properly
              self._delete_objects(
                  [
                      item["uuid"]
                      for item in insertion_errors + batch_delete_error
                  ],
                  class_name=class_name,
              )
  
      if verbose:
          self.log.info(
              "Total objects in class %s : %s ",
              class_name,
              self.conn.query.aggregate(class_name)
              .with_meta_count()
              .do(),
          )
      return insertion_errors, batch_delete_error
  
imports: from __future__ import annotations
  import contextlib
  import json
  import warnings
  from functools import cached_property
  from typing import TYPE_CHECKING, Any, Dict, List, Sequence, cast
  import requests
  import weaviate.exceptions
  from tenacity import Retrying, retry, retry_if_exception, retry_if_exception_type, stop_after_attempt
  from weaviate import Client
  from weaviate.auth import AuthApiKey, AuthBearerToken, AuthClientCredentials, AuthClientPassword
  from weaviate.data.replication import ConsistencyLevel
  from weaviate.exceptions import ObjectAlreadyExistsException
  from weaviate.util import generate_uuid5
  from airflow.exceptions import AirflowProviderDeprecationWarning
  from airflow.hooks.base import BaseHook
label_desc: Use image node_id 37 for calling the WeaviateHook obj's underlying member method code with example usage: obj.create_or_replace_document_objects(data, class_name, document_column, existing, uuid_column, vector_column, batch_config_params, tenant, verbose) and returns: insertion_errors,  batch_delete_error, list