From 1a8f07a594cd56c3ab798a93fbcc4d357ee1b5fc Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Sat, 1 Aug 2026 14:42:25 +0530 Subject: [PATCH] fix: implement WriteApi.flush() for batching mode WriteApi.flush() was documented as a stable API but was a TODO no-op. Implement it by adding a flush boundary subject that forces the current batch window to close, matching the C#/Java batch writer design. - Flush sends pending points without waiting for batch_size/flush_interval - WriteApi remains usable after flush (unlike close()) - No-op for synchronous/asynchronous write types - Unit tests for partial batch, reuse, empty buffer, and sync mode Fixes #697 --- CHANGELOG.md | 1 + influxdb_client/client/write_api.py | 39 ++++++++++++++--- tests/test_WriteApiBatching.py | 66 +++++++++++++++++++++++++++++ 3 files changed, 99 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 98a37713..8966e7ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ 1. [#706](https://github.com/influxdata/influxdb-client-python/pull/706): Use logger instead logging. 1. [#686](https://github.com/influxdata/influxdb-client-python/issues/686): Stop `default_tags` from one client leaking into other clients' `WriteApi` (shared mutable default `PointSettings`). +1. [#697](https://github.com/influxdata/influxdb-client-python/issues/697): Implement `WriteApi.flush()` to force pending batch writes without disposing the API (mirrors C#/Java batch writer). ## 1.50.0 [2026-01-23] diff --git a/influxdb_client/client/write_api.py b/influxdb_client/client/write_api.py index ea9fb7b5..f14755ee 100644 --- a/influxdb_client/client/write_api.py +++ b/influxdb_client/client/write_api.py @@ -255,13 +255,23 @@ def __init__(self, if self._write_options.write_type is WriteType.batching: # Define Subject that listen incoming data and produces writes into InfluxDB self._subject = Subject() + # Flush subject forces the current batch window to close (mirrors Java/C# clients) + self._flush_subject = Subject() self._window_scheduler = ThreadPoolScheduler(1) self._disposable = self._subject.pipe( - # Split incoming data to windows by batch_size or flush_interval - ops.window_with_time_or_count(count=write_options.batch_size, - timespan=timedelta(milliseconds=write_options.flush_interval), - scheduler=self._window_scheduler), + # Split incoming data to windows by batch_size, flush_interval, or explicit flush(). + # Window boundaries are the merge of time/count windows and the flush subject — + # same approach as the C# WriteApi batch writer. + ops.publish(lambda connected: connected.pipe( + ops.window( + rx.merge( + connected.pipe( + ops.window_with_time_or_count( + count=write_options.batch_size, + timespan=timedelta(milliseconds=write_options.flush_interval), + scheduler=self._window_scheduler)), + self._flush_subject)))), # Map window into groups defined by 'organization', 'bucket' and 'precision' ops.flat_map(lambda window: window.pipe( # Group window by 'organization', 'bucket' and 'precision' @@ -279,6 +289,7 @@ def __init__(self, else: self._subject = None + self._flush_subject = None self._disposable = None if self._write_options.write_type is WriteType.asynchronous: @@ -386,9 +397,17 @@ def write_payload(payload): return results def flush(self): - """Flush data.""" - # TODO - pass + """ + Flush data. + + Forces the client to flush all pending writes from the batching buffer to InfluxDB + via HTTP without waiting for ``batch_size`` or ``flush_interval``. The ``WriteApi`` + remains usable after ``flush()`` (unlike ``close()``). + + Has no effect when batching is not enabled (synchronous / asynchronous write type). + """ + if self._flush_subject is not None and not self._flush_subject.is_disposed: + self._flush_subject.on_next(None) def close(self): """Flush data and dispose a batching buffer.""" @@ -442,6 +461,11 @@ def __del__(self): ) break + if self._flush_subject: + self._flush_subject.on_completed() + self._flush_subject.dispose() + self._flush_subject = None + if self._window_scheduler: self._window_scheduler.executor.shutdown(wait=False) self._window_scheduler = None @@ -570,6 +594,7 @@ def __getstate__(self): state = self.__dict__.copy() # Remove rx del state['_subject'] + del state['_flush_subject'] del state['_disposable'] del state['_window_scheduler'] del state['_write_service'] diff --git a/tests/test_WriteApiBatching.py b/tests/test_WriteApiBatching.py index 8befd7e5..a569c7ad 100644 --- a/tests/test_WriteApiBatching.py +++ b/tests/test_WriteApiBatching.py @@ -736,6 +736,72 @@ def __call__(self, conf: (str, str, str), data: str, error: InfluxDBError): self.assertIsInstance(callback.error, InfluxDBError) self.assertEqual(429, callback.error.response.status) + def _wait_for_requests(self, count, timeout=5.0): + """Poll httpretty until at least ``count`` requests are recorded.""" + deadline = time.time() + timeout + while time.time() < deadline: + if len(httpretty.httpretty.latest_requests) >= count: + return + time.sleep(0.05) + self.fail("Timed out waiting for %s HTTP write(s); got %s" % ( + count, len(httpretty.httpretty.latest_requests))) + + def test_flush_partial_batch(self): + """flush() forces a write of buffered points below batch_size.""" + httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=204) + + # batch_size=2: a single point must stay buffered until flush + self._write_client.write("my-bucket", "my-org", + "h2o_feet,location=coyote_creek level\\ water_level=1 1") + + time.sleep(0.3) + self.assertEqual(0, len(httpretty.httpretty.latest_requests)) + + self._write_client.flush() + self._wait_for_requests(1) + + self.assertEqual(1, len(httpretty.httpretty.latest_requests)) + self.assertEqual("h2o_feet,location=coyote_creek level\\ water_level=1 1", + httpretty.httpretty.latest_requests[0].parsed_body) + + def test_flush_allows_reuse(self): + """WriteApi remains usable after flush() — unlike close().""" + httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=204) + + self._write_client.write("my-bucket", "my-org", + "h2o_feet,location=coyote_creek level\\ water_level=1 1") + self._write_client.flush() + self._wait_for_requests(1) + + self._write_client.write("my-bucket", "my-org", + "h2o_feet,location=coyote_creek level\\ water_level=2 2") + self._write_client.flush() + self._wait_for_requests(2) + + self.assertEqual(2, len(httpretty.httpretty.latest_requests)) + self.assertEqual("h2o_feet,location=coyote_creek level\\ water_level=1 1", + httpretty.httpretty.latest_requests[0].parsed_body) + self.assertEqual("h2o_feet,location=coyote_creek level\\ water_level=2 2", + httpretty.httpretty.latest_requests[1].parsed_body) + + def test_flush_empty_is_noop(self): + """flush() with an empty buffer does not send an HTTP request.""" + httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=204) + + self._write_client.flush() + time.sleep(0.3) + self.assertEqual(0, len(httpretty.httpretty.latest_requests)) + + def test_flush_synchronous_is_noop(self): + """flush() is a no-op for non-batching write types.""" + from influxdb_client.client.write_api import SYNCHRONOUS + + self._write_client.close() + self._write_client = WriteApi(influxdb_client=self.influxdb_client, + write_options=SYNCHRONOUS) + # Must not raise + self._write_client.flush() + if __name__ == '__main__': unittest.main()