Add `flush_after_seconds` option to `streaming_bulk()` by miguelgrinberg · Pull Request #3064 · elastic/elasticsearch-py

miguelgrinberg · 2025-09-08T16:38:16Z

This change adds a timeout controlled flush to the streaming bulk helper.

Fixes #3051

Tasks:

elasticsearch/helpers/actions.py

elasticsearch/_async/helpers.py

miguelgrinberg · 2025-09-16T15:44:02Z

test_elasticsearch/test_server/test_helpers.py

+        timestamps.append(time.time())
+
+    # make sure there is a pause between the writing of the 2nd and 3rd items
+    assert timestamps[2] - timestamps[1] > (timestamps[1] - timestamps[0]) * 2


This assert may look strange. The reason I'm validating flushing by looking at timestamp differences is to attempt to not use any fixed timings and avoid a flaky test.

pquentin

Thanks! LGTM.

pquentin · 2025-10-14T08:02:12Z

elasticsearch/compat.py

+    The thread is automatically joined when the block ends. If the thread raised
+    an exception, it is raised in the caller's context.


Out of curiosity, do you have an example traceback that we get with this?

Example test script:

import time from elasticsearch.compat import safe_thread def my_thread(): print("my_thread started") 1/0 def main(): print("main started") with safe_thread(target=my_thread): time.sleep(1) main()

Output:

❯ python threadtest.py main started my_thread started Traceback (most recent call last): File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/threadtest.py", line 14, in <module> main() ~~~~^^ File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/threadtest.py", line 10, in main with safe_thread(target=my_thread): ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^ File "/Users/miguelgrinberg/.local/share/mise/installs/python/3.14.0/lib/python3.14/contextlib.py", line 148, in __exit__ next(self.gen) ~~~~^^^^^^^^^^ File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/elasticsearch/compat.py", line 105, in safe_thread raise captured_exception File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/elasticsearch/compat.py", line 95, in run target(*args, **kwargs) ~~~~~~^^^^^^^^^^^^^^^^^ File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/s.py", line 6, in my_thread 1/0 ~^~ ZeroDivisionError: division by zero

* Add flush option to streaming_bulk() * unit tests * bulk timeouts * use context manager to run the timeout background tasks * format code * integration tests * docstrings (cherry picked from commit 6fbdecb)

* Add flush option to streaming_bulk() * unit tests * bulk timeouts * use context manager to run the timeout background tasks * format code * integration tests * docstrings (cherry picked from commit 6fbdecb) Co-authored-by: Miguel Grinberg <miguel.grinberg@gmail.com>

#3115) * Add `flush_after_seconds` option to `streaming_bulk()` (#3064) * Add flush option to streaming_bulk() * unit tests * bulk timeouts * use context manager to run the timeout background tasks * format code * integration tests * docstrings (cherry picked from commit 6fbdecb) * fix safe_task type hint for Python 3.8 --------- Co-authored-by: Miguel Grinberg <miguel.grinberg@gmail.com>

miguelgrinberg force-pushed the bulk-flush branch 3 times, most recently from cd73776 to 12762fb Compare September 9, 2025 16:05

miguelgrinberg commented Sep 9, 2025

View reviewed changes