[wip]

Author: jonavellecuerdo

tim/cli.py

@@ -361,11 +361,14 @@ def bulk_update_embeddings(
     td = TIMDEXDataset(location=dataset_path)
 
     # TODO: update TDA to read embeddings
+    records_to_update = iter()
 
     try:
-        update_results.update(tim_os.bulk_index(client, index, records_to_index))
+        update_results.update(tim_os.bulk_update(client, index, records_to_update))
     except BulkIndexingError as exception:
-        logger.info(f"Bulk indexing failed: {exception}")
+        logger.info(f"Bulk update with embeddings failed: {exception}")
+
+    logger.info(f"Bulk update with embeddings complete: {json.dumps(update_results)}")
 
 
 @main.command()

tim/opensearch.py

@@ -413,3 +413,21 @@ def bulk_index(client: OpenSearch, index: str, records: Iterator[dict]) -> dict[
     )
     logger.debug(response)
     return result
+
+
+def bulk_update(
+    client: OpenSearch, index: str, records: Iterator[dict]
+) -> dict[str, int]:
+    result = {"updated": 0, "errors": 0, "total": 0}
+    actions = helpers.generate_bulk_actions(index, records, "update")
+    responses = streaming_bulk(
+        client,
+        actions,
+        max_chunk_bytes=REQUEST_CONFIG["OPENSEARCH_BULK_MAX_CHUNK_BYTES"],
+        raise_on_error=False,
+    )
+    for response in responses:
+        # TODO: Parse different responses from bulk operations
+        pass
+    logger.debug(response)
+    return result