code: from neumai.Shared.NeumSinkInfo import NeumSinkInfo
  from neumai.Shared.NeumVector  import NeumVector
  from neumai.Shared.NeumSearch import NeumSearchResult
  from neumai.Shared.Exceptions import(
      LanceDBInsertionException,
      LanceDBIndexInfoException,
      LanceDBIndexCreationException,
      LanceDBQueryException
  )
  from neumai.SinkConnectors.SinkConnector import SinkConnector
  from typing import List, Optional
  from neumai.SinkConnectors.filter_utils import FilterCondition
  from pydantic import Field
  
  import lancedb
  from lancedb import DBConnection
  
  
  class LanceDBSink(SinkConnector):
  
      """
      LanceDB sink
  
      A sink connector for LanceDB, designed to facilitate data output into a 
      LanceDB storage system. For details about LanceDB, refer to 
      https://github.com/lancedb/lancedb.
  
      LanceDB supports flat search as well as ANN search.
      For indexing, read here - https://lancedb.github.io/lancedb/ann_indexes/#creating-an-ivf_pq-index
  
      Attributes:
      -----------
      uri: str
          URI for LanceDB database.
      api_key: str
          If presented, connect to LanceDB cloud. 
          Otherwise, connect to a database on file system or cloud storage.
      region: str
          Region for use of LanceDB cloud.
      table_name: str
          Name of LanceDB table to use
      create_index: bool
          LanceDB offers flat search as well as ANN search. If set to True,
          a vector index would be created for searching instead of a 
          brute-force knn search.
      metric: str
          The distance metric to use. By default it uses euclidean distance 'L2'. 
          It also supports 'cosine' and 'dot' distance as well. Needs to be set if create_index is True.
      num_partitions: int
          The number of partitions of the index. 
          Needs to be set if create_index is True. And needs to be altered as per data size.
      num_sub_vectors: int
          The number of sub-vectors (M) that will be created during 
          Product Quantization (PQ). For D dimensional vector, it will be divided into 
          M of D/M sub-vectors, each of which is presented by a single PQ code.
      accelerator: str
          The accelerator to use for the index creation process. Supports GPU and MPS.
  
  
      Example usage:
          ldb = LanceDBSink(uri="data/test_ldb_sink", table_name="demo_ldb_table")
          ldb.store(neum_vectors)
          ldb.search(query)
      """
  
      uri: str = Field(..., description="URI for LanceDB database")
      api_key: Optional[str] = Field(default=None, description="API key for LanceDB cloud")
      region: Optional[str] = Field(default=None, description="Region for use of LanceDB cloud")
      table_name: str = Field(..., description="Name of LanceDB table to use")
      create_index: bool = Field(default=False, description="Boolean to create index or use flat search")
      metric: str = Field(default="cosine", description="The distance metric to use in the index")
      num_partitions: int = Field(default=256, description="The number of partitions of the index")
      num_sub_vectors: int = Field(default=96, description="The number of sub-vectors (M) that will be created during Product Quantization (PQ)")
      accelerator: str = Field(default=None, description="Specify to cuda or mps (on Apple Silicon) to enable GPU training.")
  
      # Check API reference for more details
      # - https://lancedb.github.io/lancedb/python/python/#lancedb.connect
      # db: DBConnection = lancedb.connect(uri=uri, api_key=api_key, region=region)
  
      @property
      def sink_name(self) -> str:
          return "LanceDBSink"
      
      @property
      def required_properties(self) -> List[str]:
          return ['uri', 'api_key', 'table_name']
      
      @property
      def optional_properties(self) -> List[str]:
          return []
      
      def validation(self) -> bool:
          """config_validation connector setup"""
          db = lancedb.connect(uri=self.uri, api_key=self.api_key, region=self.region)
          return True 
      
      def _get_db_connection(self) -> DBConnection:
          return lancedb.connect(uri=self.uri, api_key=self.api_key, region=self.region)
  
      def store(self, vectors_to_store: List[NeumVector]) -> int:
          db = self._get_db_connection()
          table_name = self.table_name
  
          data = []
          for vec in vectors_to_store:
              dic = {
                  'id': vec.id,
                  'vector': vec.vector,
              }
              for k,v in vec.metadata.items():
                  dic[k] = v
              data.append(dic)
  
          tbl = db.create_table(table_name, data=data, mode="overwrite")
          if tbl:
              return len(tbl.to_pandas())
          raise LanceDBInsertionException("LanceDB storing failed. Try later")
      
  
      def search(self, vector: List[float], number_of_results: int, filters: List[FilterCondition] = []) -> List[NeumSearchResult]:
  
          db = self._get_db_connection()
          tbl = db.open_table(self.table_name)
  
          if self.create_index:
              # For more details, refer to docs
              # - https://lancedb.github.io/lancedb/python/python/#lancedb.table.Table.create_index
              try:
                  tbl.create_index(
                      metric=self.metric, 
                      num_partitions=self.num_partitions,
                      num_sub_vectors=self.num_sub_vectors,
                      accelerator=self.accelerator,
                      replace=True)
              except Exception as e:
                  raise LanceDBIndexCreationException(f"LanceDB index creation failed. \nException - {e}")
  
          try:
              search_results = tbl.search(query=vector)
              for filter in filters:
                  search_results = search_results.where(f"{filter.field} {filter.operator.value} {filter.value}")
              search_results = search_results.limit(number_of_results).to_pandas()
  
          except Exception as e:
              raise LanceDBQueryException(f"Failed to query LanceDB. Exception - {e}")
  
          matches = []
          cols = search_results.columns
  
          for i in range(len(search_results)):
              _id = search_results.iloc[i]['id']
              _vec = list(search_results.iloc[i]['vector'])
              matches.append(
                  NeumSearchResult(
                      id=_id,
                      vector=_vec,
                      metadata={k:search_results.iloc[i][k] for k in cols if k not in ['id', 'vector', '_distance']},
                      score=1-search_results.iloc[i]['_distance']
                  )
              )
          return matches
      
  
      def get_representative_vector(self) -> list:
          db = self._get_db_connection()
          tbl = db.open_table(self.table_name)
          return list(tbl.to_pandas()['vector'].mean())
      
      
      def info(self) -> NeumSinkInfo:
          try:
              db = self._get_db_connection()
              tbl = db.open_table(self.table_name)
              return(NeumSinkInfo(number_vectors_stored=len(tbl)))
          except Exception as e:
              raise LanceDBIndexInfoException(f"Failed to get information from LanceDB. Exception - {e}")
          
  
      def delete_vectors_with_file_id(self, file_id: str) -> bool:
          db = self._get_db_connection()
          table_name = self.table_name
  
          tbl = db.open_table(table_name)
          try:
              tbl.delete(where=f"id = '{file_id}'")
          except:
              raise Exception("LanceDB deletion by file id failed.")
          return True
apis: lancedb.connect
extract_api: [((2397, 2447), 'pydantic.Field', 'Field', (['...'], {'description': '"""URI for LanceDB database"""'}), "(..., description='URI for LanceDB database')\n", (2402, 2447), False, 'from pydantic import Field\n'), ((2477, 2537), 'pydantic.Field', 'Field', ([], {'default': 'None', 'description': '"""API key for LanceDB cloud"""'}), "(default=None, description='API key for LanceDB cloud')\n", (2482, 2537), False, 'from pydantic import Field\n'), ((2566, 2632), 'pydantic.Field', 'Field', ([], {'default': 'None', 'description': '"""Region for use of LanceDB cloud"""'}), "(default=None, description='Region for use of LanceDB cloud')\n", (2571, 2632), False, 'from pydantic import Field\n'), ((2655, 2709), 'pydantic.Field', 'Field', (['...'], {'description': '"""Name of LanceDB table to use"""'}), "(..., description='Name of LanceDB table to use')\n", (2660, 2709), False, 'from pydantic import Field\n'), ((2735, 2813), 'pydantic.Field', 'Field', ([], {'default': '(False)', 'description': '"""Boolean to create index or use flat search"""'}), "(default=False, description='Boolean to create index or use flat search')\n", (2740, 2813), False, 'from pydantic import Field\n'), ((2832, 2910), 'pydantic.Field', 'Field', ([], {'default': '"""cosine"""', 'description': '"""The distance metric to use in the index"""'}), "(default='cosine', description='The distance metric to use in the index')\n", (2837, 2910), False, 'from pydantic import Field\n'), ((2937, 3008), 'pydantic.Field', 'Field', ([], {'default': '(256)', 'description': '"""The number of partitions of the index"""'}), "(default=256, description='The number of partitions of the index')\n", (2942, 3008), False, 'from pydantic import Field\n'), ((3036, 3162), 'pydantic.Field', 'Field', ([], {'default': '(96)', 'description': '"""The number of sub-vectors (M) that will be created during Product Quantization (PQ)"""'}), "(default=96, description=\n    'The number of sub-vectors (M) that will be created during Product Quantization (PQ)'\n    )\n", (3041, 3162), False, 'from pydantic import Field\n'), ((3176, 3281), 'pydantic.Field', 'Field', ([], {'default': 'None', 'description': '"""Specify to cuda or mps (on Apple Silicon) to enable GPU training."""'}), "(default=None, description=\n    'Specify to cuda or mps (on Apple Silicon) to enable GPU training.')\n", (3181, 3281), False, 'from pydantic import Field\n'), ((3852, 3923), 'lancedb.connect', 'lancedb.connect', ([], {'uri': 'self.uri', 'api_key': 'self.api_key', 'region': 'self.region'}), '(uri=self.uri, api_key=self.api_key, region=self.region)\n', (3867, 3923), False, 'import lancedb\n'), ((4015, 4086), 'lancedb.connect', 'lancedb.connect', ([], {'uri': 'self.uri', 'api_key': 'self.api_key', 'region': 'self.region'}), '(uri=self.uri, api_key=self.api_key, region=self.region)\n', (4030, 4086), False, 'import lancedb\n'), ((4629, 4691), 'neumai.Shared.Exceptions.LanceDBInsertionException', 'LanceDBInsertionException', (['"""LanceDB storing failed. Try later"""'], {}), "('LanceDB storing failed. Try later')\n", (4654, 4691), False, 'from neumai.Shared.Exceptions import LanceDBInsertionException, LanceDBIndexInfoException, LanceDBIndexCreationException, LanceDBQueryException\n'), ((5864, 5930), 'neumai.Shared.Exceptions.LanceDBQueryException', 'LanceDBQueryException', (['f"""Failed to query LanceDB. Exception - {e}"""'], {}), "(f'Failed to query LanceDB. Exception - {e}')\n", (5885, 5930), False, 'from neumai.Shared.Exceptions import LanceDBInsertionException, LanceDBIndexInfoException, LanceDBIndexCreationException, LanceDBQueryException\n'), ((6186, 6376), 'neumai.Shared.NeumSearch.NeumSearchResult', 'NeumSearchResult', ([], {'id': '_id', 'vector': '_vec', 'metadata': "{k: search_results.iloc[i][k] for k in cols if k not in ['id', 'vector',\n    '_distance']}", 'score': "(1 - search_results.iloc[i]['_distance'])"}), "(id=_id, vector=_vec, metadata={k: search_results.iloc[i][k\n    ] for k in cols if k not in ['id', 'vector', '_distance']}, score=1 -\n    search_results.iloc[i]['_distance'])\n", (6202, 6376), False, 'from neumai.Shared.NeumSearch import NeumSearchResult\n'), ((6958, 7048), 'neumai.Shared.Exceptions.LanceDBIndexInfoException', 'LanceDBIndexInfoException', (['f"""Failed to get information from LanceDB. Exception - {e}"""'], {}), "(\n    f'Failed to get information from LanceDB. Exception - {e}')\n", (6983, 7048), False, 'from neumai.Shared.Exceptions import LanceDBInsertionException, LanceDBIndexInfoException, LanceDBIndexCreationException, LanceDBQueryException\n'), ((5435, 5525), 'neumai.Shared.Exceptions.LanceDBIndexCreationException', 'LanceDBIndexCreationException', (['f"""LanceDB index creation failed. \nException - {e}"""'], {}), '(\n    f"""LanceDB index creation failed. \nException - {e}""")\n', (5464, 5525), False, 'from neumai.Shared.Exceptions import LanceDBInsertionException, LanceDBIndexInfoException, LanceDBIndexCreationException, LanceDBQueryException\n')]