feat(api): support bulk elasticsearch operations

api/howler/api/v1/hit.py

@@ -35,6 +35,7 @@
 from howler.odm.models.user import User
 from howler.security import api_login
 from howler.services import action_service, analytic_service, event_service, hit_service
+from howler.utils.constants import DEBUG_FORCE_REFRESH
 from howler.utils.str_utils import sanitize_lucene_query
 
 MAX_COMMENT_LEN = 5000
@@ -125,17 +126,12 @@ def create_hits(user: User, **kwargs):
 
     if len(response_body["invalid"]) == 0:
         if len(odms) > 0:
-            for odm in odms:  # Save all but the last hit without refreshing
+            for odm in odms:
                 # Ensure all ids are consistent
                 if odm.event is not None:
                     odm.event.id = odm.howler.id
-                # TODO keeping the commit approach until we can batch create hits
-                # passing refresh to each one is potentially even more inefficient
-                hit_service.create_hit(odm.howler.id, odm, user=user.uname, refresh="false")
-
-            if refresh == "true":
-                datastore().hit.commit()
 
+            hit_service.create_hits(odms, user=user.uname, refresh="true" if DEBUG_FORCE_REFRESH else refresh)
             analytic_service.save_from_hits(odms, user, refresh=refresh)
             action_service.enqueue_action_execution([odm.howler.id for odm in odms], trigger="create", user=user)
 
@@ -589,7 +585,7 @@ def add_label(id, label_set, user, **kwargs):
         id,
         [hit_helper.list_add(f"howler.labels.{label_set}", label) for label in labels],
         user["uname"],
-        refresh=refresh,
+        refresh="true" if DEBUG_FORCE_REFRESH else refresh,
     )
 
     action_service.enqueue_action_execution([id], trigger="add_label", user=user)
@@ -646,7 +642,7 @@ def remove_labels(id, label_set, user, **kwargs):
         id,
         [hit_helper.list_remove(f"howler.labels.{label_set}", label) for label in labels],
         user["uname"],
-        refresh=refresh,
+        refresh="true" if DEBUG_FORCE_REFRESH else refresh,
     )
 
     action_service.enqueue_action_execution([id], trigger="remove_label", user=user)
@@ -1078,23 +1074,22 @@ def create_bundle(user: User, **kwargs):
 
         odm.howler.hits.extend(hit_id for hit_id in child_hits if hit_id not in odm.howler.hits)
 
-        # TODO see comment above about batch creating hits
-        hit_service.create_hit(odm.howler.id, odm, user=user.uname, refresh="false")
+        hit_service.create_hit(odm.howler.id, odm, user=user.uname, refresh=refresh)
         analytic_service.save_from_hits(odm, user, refresh=refresh)
 
+        child_hits = []
         for hit_id in odm.howler.hits:
             child_hit: Hit = hit_service.get_hit(hit_id, as_odm=True)
+            child_hits.append(child_hit)
 
             if child_hit.howler.is_bundle:
                 return bad_request(
                     err=f"You cannot specify a bundle as a child of another bundle - {child_hit.howler.id} is a bundle."
                 )
 
             child_hit.howler.bundles.append(odm.howler.id)
-            datastore().hit.save(child_hit.howler.id, child_hit, refresh="false")
 
-        if refresh == "true":
-            datastore().hit.commit()
+        hit_service.overwrite_hits(child_hits, refresh=refresh)
 
         return created(odm)
     except HowlerException as e:
@@ -1152,8 +1147,10 @@ def update_bundle(id, **kwargs):
     bundle_hit.howler.is_bundle = True
 
     try:
+        hits_to_save = []
         for hit_id in new_hit_list:
             child_hit: Hit = hit_service.get_hit(hit_id, as_odm=True)
+            hits_to_save.append(child_hit)
 
             if child_hit.howler.is_bundle:
                 return bad_request(
@@ -1163,9 +1160,9 @@ def update_bundle(id, **kwargs):
             new_bundle_list = child_hit.howler.as_primitives().get("bundles", [])
             new_bundle_list.append(bundle_hit.howler.id)
             child_hit.howler.bundles = new_bundle_list
-            datastore().hit.save(child_hit.howler.id, child_hit)
 
-        datastore().hit.save(bundle_hit.howler.id, bundle_hit, refresh=refresh)
+        hits_to_save.append(bundle_hit)
+        hit_service.overwrite_hits(hits_to_save, refresh=refresh)
 
         return ok(bundle_hit)
     except HowlerException as e:
@@ -1226,17 +1223,18 @@ def remove_bundle_children(id, **kwargs):
     bundle_hit.howler.is_bundle = len(new_hit_list) > 0
 
     try:
+        hits_to_save = []
         for hit_id in hit_ids:
             child_hit: Hit = hit_service.get_hit(hit_id, as_odm=True)
+            hits_to_save.append(child_hit)
 
             try:
                 child_hit.howler.bundles.remove(bundle_hit.howler.id)
             except ValueError:
                 logger.warning("Bundle isn't included in child %s!", bundle_hit.howler.id)
 
-            datastore().hit.save(child_hit.howler.id, child_hit)
-
-        datastore().hit.save(bundle_hit.howler.id, bundle_hit, refresh=refresh)
+        hits_to_save.append(bundle_hit)
+        hit_service.overwrite_hits(hits_to_save, refresh=refresh)
 
         return ok(bundle_hit)
     except HowlerException as e:

api/howler/api/v1/tool.py

@@ -3,6 +3,7 @@
 from flask import request
 
 from howler.api import bad_request, created, make_subapi_blueprint
+from howler.api.v1.utils.params import parse_parameters, parse_refresh
 from howler.common.exceptions import HowlerException
 from howler.common.loader import datastore
 from howler.common.logging import get_logger
@@ -13,6 +14,7 @@
 from howler.odm.models.user import User
 from howler.security import api_login
 from howler.services import action_service, analytic_service, hit_service
+from howler.utils.constants import DEBUG_FORCE_REFRESH
 from howler.utils.dict_utils import flatten
 from howler.utils.isotime import now_as_iso
 from howler.utils.str_utils import get_parent_key
@@ -30,6 +32,7 @@
 @generate_swagger_docs()
 @tool_api.route("/<tool_name>/hits", methods=["POST", "PUT"])
 @api_login(required_priv=["W"])
+@parse_parameters(refresh=parse_refresh)
 def create_one_or_many_hits(tool_name: str, user: User, **kwargs):  # noqa: C901
     """Create one or many hits for a tool using field mapping.
 
@@ -64,6 +67,7 @@ def create_one_or_many_hits(tool_name: str, user: User, **kwargs):  # noqa: C901
         Use POST /api/v1/hit/ directly with pre-mapped hit data instead.
     """
     data = request.json
+    refresh = kwargs.get("refresh", None)
     if not isinstance(data, dict):
         return bad_request(err="Invalid data format")
 
@@ -194,17 +198,11 @@ def create_one_or_many_hits(tool_name: str, user: User, **kwargs):  # noqa: C901
             bundle_hit.howler.bundle_size += 1
             odm.howler.bundles.append(bundle_hit.howler.id)
 
-        hit_service.create_hit(odm.howler.id, odm, user=user.uname)
-
-    analytic_service.save_from_hits(odms, user)
-
     if bundle_hit:
-        hit_service.create_hit(bundle_hit.howler.id, bundle_hit, user=user.uname)
-
-        analytic_service.save_from_hits(bundle_hit, user)
-
-    datastore().hit.commit()
+        odms.append(bundle_hit)
 
+    hit_service.create_hits(odms, user=user.uname, refresh="true" if DEBUG_FORCE_REFRESH else refresh)
+    analytic_service.save_from_hits(odms, user, refresh=refresh)
     action_service.enqueue_action_execution([entry["id"] for entry in out], trigger="create", user=user)
 
     return created(out, warnings=warnings)

api/howler/datastore/bulk.py

@@ -1,72 +1,145 @@
 import json
+import warnings
 from copy import deepcopy
+from typing import List, Optional
+
+from howler import odm
+
+_OPERATION_GROUP = tuple[str] | tuple[str, str]
 
 
 class ElasticBulkPlan(object):
-    def __init__(self, indexes, model=None):
+    ELASTIC_MAX_REQUEST_SIZE = 100_000_000  # 100MB
+    DEFAULT_BATCH_SIZE = 500
+
+    def __init__(self, indexes: List[str], model: Optional[type[odm.Model]] = None):
         self.indexes = indexes
         self.model = model
-        self.operations = []
+        self.operations: list[_OPERATION_GROUP] = []
 
     @property
     def empty(self):
         return len(self.operations) == 0
 
     def add_delete_operation(self, doc_id, index=None):
         if index:
-            self.operations.append(json.dumps({"delete": {"_index": index, "_id": doc_id}}))
+            self.operations.append((json.dumps({"delete": {"_index": index, "_id": doc_id}}),))
         else:
             for cur_index in self.indexes:
-                self.operations.append(json.dumps({"delete": {"_index": cur_index, "_id": doc_id}}))
+                self.operations.append((json.dumps({"delete": {"_index": cur_index, "_id": doc_id}}),))
+
+    def add_insert_operation(self, doc_id: str, doc, index=None):
+        if self.model and isinstance(doc, self.model):
+            saved_doc = doc.as_primitives(hidden_fields=True)
+        elif self.model:
+            saved_doc = self.model(doc).as_primitives(hidden_fields=True)
+        else:
+            if not isinstance(doc, dict):
+                saved_doc = {"__non_doc_raw__": doc}
+            else:
+                saved_doc = deepcopy(doc)
+        saved_doc["id"] = doc_id
 
-    def add_insert_operation(self, doc_id, doc, index=None):
-        if isinstance(doc, self.model):
-            saved_doc = doc.as_primitives(hidden_fields=True)  # type: ignore
+        self.operations.append(
+            (
+                json.dumps({"create": {"_index": index or self.indexes[0], "_id": doc_id}}),
+                json.dumps(saved_doc),
+            )
+        )
+
+    def add_index_operation(self, doc_id, doc, index=None):
+        if self.model and isinstance(doc, self.model):
+            saved_doc = doc.as_primitives(hidden_fields=True)
         elif self.model:
-            saved_doc = self.model(doc).as_primitives(hidden_fields=True)  # type: ignore
+            saved_doc = self.model(doc).as_primitives(hidden_fields=True)
         else:
             if not isinstance(doc, dict):
                 saved_doc = {"__non_doc_raw__": doc}
             else:
-                saved_doc = deepcopy(doc)  # type: ignore
+                saved_doc = deepcopy(doc)
         saved_doc["id"] = doc_id
 
-        self.operations.append(json.dumps({"create": {"_index": index or self.indexes[0], "_id": doc_id}}))
-        self.operations.append(json.dumps(saved_doc))
+        self.operations.append(
+            (
+                json.dumps({"index": {"_index": index or self.indexes[0], "_id": doc_id}}),
+                json.dumps(saved_doc),
+            )
+        )
 
     def add_upsert_operation(self, doc_id, doc, index=None):
-        if isinstance(doc, self.model):
-            saved_doc = doc.as_primitives(hidden_fields=True)  # type: ignore
+        if self.model and isinstance(doc, self.model):
+            saved_doc = doc.as_primitives(hidden_fields=True)
         elif self.model:
-            saved_doc = self.model(doc).as_primitives(hidden_fields=True)  # type: ignore
+            saved_doc = self.model(doc).as_primitives(hidden_fields=True)
         else:
             if not isinstance(doc, dict):
-                saved_doc = {"__non_doc_raw__": doc}  # type: ignore
+                saved_doc = {"__non_doc_raw__": doc}
             else:
-                saved_doc = deepcopy(doc)  # type: ignore
+                saved_doc = deepcopy(doc)
         saved_doc["id"] = doc_id
 
-        self.operations.append(json.dumps({"update": {"_index": index or self.indexes[0], "_id": doc_id}}))
-        self.operations.append(json.dumps({"doc": saved_doc, "doc_as_upsert": True}))
+        self.operations.append(
+            (
+                json.dumps({"update": {"_index": index or self.indexes[0], "_id": doc_id}}),
+                json.dumps({"doc": saved_doc, "doc_as_upsert": True}),
+            )
+        )
 
     def add_update_operation(self, doc_id, doc, index=None):
-        if isinstance(doc, self.model):
-            saved_doc = doc.as_primitives(hidden_fields=True)  # type: ignore
+        if self.model and isinstance(doc, self.model):
+            saved_doc = doc.as_primitives(hidden_fields=True)
         elif self.model:
-            saved_doc = self.model(doc, mask=list(doc.keys())).as_primitives(hidden_fields=True)  # type: ignore
+            saved_doc = self.model(doc, mask=list(doc.keys())).as_primitives(hidden_fields=True)
         else:
             if not isinstance(doc, dict):
-                saved_doc = {"__non_doc_raw__": doc}  # type: ignore
+                saved_doc = {"__non_doc_raw__": doc}
             else:
-                saved_doc = deepcopy(doc)  # type: ignore
+                saved_doc = deepcopy(doc)
 
         if index:
-            self.operations.append(json.dumps({"update": {"_index": index, "_id": doc_id}}))
-            self.operations.append(json.dumps({"doc": saved_doc}))
+            self.operations.append(
+                (
+                    json.dumps({"update": {"_index": index, "_id": doc_id}}),
+                    json.dumps({"doc": saved_doc}),
+                )
+            )
         else:
             for cur_index in self.indexes:
-                self.operations.append(json.dumps({"update": {"_index": cur_index, "_id": doc_id}}))
-                self.operations.append(json.dumps({"doc": saved_doc}))
+                self.operations.append(
+                    (
+                        json.dumps({"update": {"_index": cur_index, "_id": doc_id}}),
+                        json.dumps({"doc": saved_doc}),
+                    )
+                )
 
     def get_plan_data(self):
-        return "\n".join(self.operations)
+        """Construct the bulk request from the current operations"""
+        return self._get_plan_for_operations()
+
+    def get_plan_batches(self, batch_size: int = DEFAULT_BATCH_SIZE):
+        """Yield plan data in batches"""
+        for ptr in range(0, len(self.operations), batch_size):
+            yield self._get_plan_for_operations(self.operations[ptr : ptr + batch_size])
+
+    def _flatten_operations(self, batch: List[_OPERATION_GROUP] | None = None) -> List[str]:
+        """Flatten the (operation, data) tuples for a batch or the full list of operations if no batch provided"""
+        if not batch:
+            batch = self.operations
+
+        flattened: list[str] = []
+        for op in batch:
+            flattened.extend(op)
+
+        return flattened
+
+    def _get_plan_for_operations(self, batch: List[_OPERATION_GROUP] | None = None) -> str:
+        """Get the bulk plan string for a batch or the full list of operations if no batch provided"""
+        plan = "\n".join(self._flatten_operations(batch)) + "\n"
+
+        if len(plan.encode("utf-8")) > self.ELASTIC_MAX_REQUEST_SIZE:
+            warnings.warn(
+                f"Bulk plan exceeds maximum request size of {self.ELASTIC_MAX_REQUEST_SIZE} bytes. "
+                f"Current size: {len(plan.encode('utf-8'))} bytes."
+            )
+
+        return plan

api/howler/datastore/collection.py

@@ -22,6 +22,7 @@
 from howler.common.exceptions import HowlerRuntimeError, HowlerValueError, NonRecoverableError
 from howler.common.loader import APP_NAME
 from howler.common.logging.format import HWL_DATE_FORMAT, HWL_LOG_FORMAT
+from howler.datastore.bulk import ElasticBulkPlan
 from howler.datastore.constants import BACK_MAPPING, TYPE_MAPPING
 from howler.datastore.exceptions import (
     DataStoreException,
@@ -532,6 +533,26 @@ def _update_async(self, index, script, query, max_docs=None, refresh=None):
             else:
                 updated += res["updated"]
 
+    def bulk(self, operations: ElasticBulkPlan, refresh: str | None = None):
+        """
+        Execute a bulk plan.
+
+        :return: True if the operation completed without errors
+        """
+        responses = []
+        for operation_batch in operations.get_plan_batches():
+            response = self.with_retries(self.datastore.client.bulk, operations=operation_batch, refresh=refresh)
+            responses.append(response)
+        return not any(response["errors"] for response in responses)
+
+    def get_bulk_plan(self):
+        """
+        Create a BulkPlan tailored for the current datastore
+
+        :return: The BulkPlan object
+        """
+        return ElasticBulkPlan(self.index_list, self.model_class)
+
     @tracer.start_as_current_span(f"{__name__}.commit")
     def commit(self):
         """This function should be overloaded to perform a commit of the index data of all the different hosts

api/howler/services/analytic_service.py

@@ -119,11 +119,11 @@ def save_from_hits(hits: Hit | list[Hit], user: User, refresh: str | None = None
             analytics.append(analytic_update)
 
     if analytics:
-        for analytic in analytics[:-1]:
-            storage.analytic.save(analytic.analytic_id, analytic)
+        bulk_plan = storage.analytic.get_bulk_plan()
+        for analytic in analytics:
+            bulk_plan.add_index_operation(analytic.analytic_id, analytic)
 
-        # save the last one passing on the refresh parameter
-        storage.analytic.save(analytics[-1].analytic_id, analytics[-1], refresh=refresh)
+        storage.analytic.bulk(bulk_plan, refresh=refresh)
 
 
 def _get_analytic_updates_from_hit_group(analytic_name: str, hit_group: list[Hit], user: User) -> Analytic | None: