feat(api): support bulk elasticsearch operations#470
Open
cccs-hxp wants to merge 24 commits into
Open
Conversation
…or complex actions
Contributor
There was a problem hiding this comment.
Pull request overview
Adds bulk-write support to Howler’s Elasticsearch datastore layer and updates hit/analytic write paths to use bulk operations for multi-document workflows, with integration tests validating refresh forwarding behavior.
Changes:
- Introduce
ElasticBulkPlanand addESCollection.bulk()/ESCollection.get_bulk_plan()to execute bulk plans. - Update hit and tool APIs/services to create/overwrite multiple hits via bulk requests and update analytics via a single bulk request.
- Expand/adjust integration tests to cover bulk behavior and ensure
refreshis forwarded correctly across multi-write endpoints.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| api/howler/datastore/bulk.py | Adds typing improvements to ElasticBulkPlan used to build bulk NDJSON operations. |
| api/howler/datastore/collection.py | Adds bulk() execution and get_bulk_plan() factory on ESCollection. |
| api/howler/services/analytic_service.py | Switches analytic updates from per-doc saves to a bulk upsert plan. |
| api/howler/services/hit_service.py | Adds bulk create_hits and overwrite_hits helpers. |
| api/howler/api/v1/hit.py | Uses bulk hit creation and bulk overwrite for bundle-related multi-hit writes; adds debug refresh forcing. |
| api/howler/api/v1/tool.py | Parses refresh and uses bulk hit creation for tool hit ingestion; adds debug refresh forcing. |
| api/howler/utils/constants.py | Introduces DEBUG_FORCE_REFRESH toggle (currently defaulting to enabled). |
| api/test/integration/test_datastore.py | Adds an integration test exercising bulk delete/insert plans. |
| api/test/integration/api/test_wait_for_refresh.py | Refactors refresh-forwarding tests using spy wrappers to validate refresh across multi-index/multi-write endpoints. |
Contributor
Howler API - Coverage Results
Diff CoverageDiff: origin/main...HEAD, staged and unstaged changes
Summary
api/howler/datastore/bulk.pyLines 23-31 23 return len(self.operations) == 0
24
25 def add_delete_operation(self, doc_id, index=None):
26 if index:
! 27 self.operations.append((json.dumps({"delete": {"_index": index, "_id": doc_id}}),))
28 else:
29 for cur_index in self.indexes:
30 self.operations.append((json.dumps({"delete": {"_index": cur_index, "_id": doc_id}}),))Lines 32-43 32 def add_insert_operation(self, doc_id: str, doc, index=None):
33 if self.model and isinstance(doc, self.model):
34 saved_doc = doc.as_primitives(hidden_fields=True)
35 elif self.model:
! 36 saved_doc = self.model(doc).as_primitives(hidden_fields=True)
37 else:
38 if not isinstance(doc, dict):
! 39 saved_doc = {"__non_doc_raw__": doc}
40 else:
41 saved_doc = deepcopy(doc)
42 saved_doc["id"] = doc_idLines 51-59 51 def add_index_operation(self, doc_id, doc, index=None):
52 if self.model and isinstance(doc, self.model):
53 saved_doc = doc.as_primitives(hidden_fields=True)
54 elif self.model:
! 55 saved_doc = self.model(doc).as_primitives(hidden_fields=True)
56 else:
57 if not isinstance(doc, dict):
58 saved_doc = {"__non_doc_raw__": doc}
59 else:Lines 67-86 67 )
68 )
69
70 def add_upsert_operation(self, doc_id, doc, index=None):
! 71 if self.model and isinstance(doc, self.model):
! 72 saved_doc = doc.as_primitives(hidden_fields=True)
73 elif self.model:
! 74 saved_doc = self.model(doc).as_primitives(hidden_fields=True)
75 else:
76 if not isinstance(doc, dict):
! 77 saved_doc = {"__non_doc_raw__": doc}
78 else:
! 79 saved_doc = deepcopy(doc)
80 saved_doc["id"] = doc_id
81
! 82 self.operations.append(
83 (
84 json.dumps({"update": {"_index": index or self.indexes[0], "_id": doc_id}}),
85 json.dumps({"doc": saved_doc, "doc_as_upsert": True}),
86 )Lines 87-105 87 )
88
89 def add_update_operation(self, doc_id, doc, index=None):
90 if self.model and isinstance(doc, self.model):
! 91 saved_doc = doc.as_primitives(hidden_fields=True)
92 elif self.model:
! 93 saved_doc = self.model(doc, mask=list(doc.keys())).as_primitives(hidden_fields=True)
94 else:
95 if not isinstance(doc, dict):
! 96 saved_doc = {"__non_doc_raw__": doc}
97 else:
98 saved_doc = deepcopy(doc)
99
100 if index:
! 101 self.operations.append(
102 (
103 json.dumps({"update": {"_index": index, "_id": doc_id}}),
104 json.dumps({"doc": saved_doc}),
105 )Lines 139-147 139 """Get the bulk plan string for a batch or the full list of operations if no batch provided"""
140 plan = "\n".join(self._flatten_operations(batch)) + "\n"
141
142 if ELASTIC_MAX_REQUEST_SIZE and len(plan.encode("utf-8")) > ELASTIC_MAX_REQUEST_SIZE:
! 143 warnings.warn(
144 f"Bulk plan exceeds maximum request size of {ELASTIC_MAX_REQUEST_SIZE} bytes. "
145 f"Current size: {len(plan.encode('utf-8'))} bytes."
146 )api/howler/services/hit_service.pyLines 443-451 443 bulk_plan = storage.hit.get_bulk_plan()
444
445 for hit in hits:
446 if not overwrite and storage.hit.exists(hit.howler.id):
! 447 raise ResourceExists("Hit %s already exists in datastore" % hit.howler.id)
448
449 if user:
450 hit.howler.log = [Log({"timestamp": "NOW", "explanation": "Created hit", "user": user})]Lines 494-508 494 """Bulk save multiple hits to the datastore.
495
496 Similar to save_hit for batch without versioning. Will overwrite existing hits with the same ID.
497 """
! 498 storage = datastore()
! 499 bulk_plan = storage.hit.get_bulk_plan()
500
! 501 for hit in hits:
! 502 bulk_plan.add_index_operation(hit.howler.id, hit)
503
! 504 return storage.hit.bulk(bulk_plan, refresh=refresh)
505
506
507 @typing.no_type_check
508 @tracer.start_as_current_span(f"{__name__}.save_hit")Full Coverage ReportExpand |
-passing-brightgreen){kind=icon}
cccs-hxp
commented
Jul 6, 2026
cccs-mdr
requested changes
Jul 7, 2026
cccs-mdr
left a comment
cccs-mdr
left a comment
Collaborator
There was a problem hiding this comment.
Mostly LGTM, a few comments.
cccs-mdr
approved these changes
Jul 10, 2026
Overwrite test_wait_for_refresh because tests have changed significantly
Contributor
Author
|
@cccs-mdr I think I can add bulk versions for each of the current case append functions (extend_ ?) but it wouldn't use the mixin save method. What do you think? Do we want the ability to update a case's items in bulk? |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add functionality to our interface to elasticsearch to support bulk operations through the ElasticBulkPlan class and use a bulk request for endpoints that load or update multiple hits.
Changes: