fix: implement WriteApi.flush() for batching mode

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]
 

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']

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()