diff --git a/CHANGELOG.md b/CHANGELOG.md
index ca6775c6..5362caf6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,7 @@
 1. [#714](https://github.com/influxdata/influxdb-client-python/pull/714): Implement `WriteApi.flush()` to force pending batch writes without disposing the API (mirrors C#/Java batch writer).
 1. [#715](https://github.com/influxdata/influxdb-client-python/pull/715): Accept numeric zero for `RangeThreshold.min`/`max` and greater/lesser `value` (identity checks vs `None`, not truthiness); cover create-check serialization.
 1. [#720](https://github.com/influxdata/influxdb-client-python/pull/720): Prevent the `api_client` class from raising exceptions while the Python interpreter clears the client state during shutdown.
+1. [#733](https://github.com/influxdata/influxdb-client-python/pull/733): Write the last batch when the batching `WriteApi` is closed in an `atexit` handler instead of dropping it with "cannot schedule new futures after interpreter shutdown".
 
 ## 1.50.0 [2026-01-23]
 
diff --git a/influxdb_client/client/write_api.py b/influxdb_client/client/write_api.py
index 67892224..21c93e4e 100644
--- a/influxdb_client/client/write_api.py
+++ b/influxdb_client/client/write_api.py
@@ -194,6 +194,11 @@ def _body_reduce(batch_items):
     return b'\n'.join(map(lambda batch_item: batch_item.data, batch_items))
 
 
+def _is_interpreter_shutdown(exception: Exception) -> bool:
+    # e.g. "cannot schedule new futures after interpreter shutdown"
+    return isinstance(exception, RuntimeError) and 'interpreter shutdown' in str(exception)
+
+
 class WriteApi(_BaseWriteApi):
     """
     Implementation for '/api/v2/write' endpoint.
@@ -548,6 +553,10 @@ def _to_response(self, data: _BatchItem, delay: timedelta):
             ops.subscribe_on(self._write_options.write_scheduler),
             # use delay if its specified
             ops.delay(duetime=delay, scheduler=self._write_options.write_scheduler),
+            # since Python 3.9 the thread pools stop accepting work before 'atexit' handlers run,
+            # so write the batch on the current thread instead of dropping it
+            ops.catch(handler=lambda exception, source: rx.just(data) if _is_interpreter_shutdown(exception)
+                      else rx.throw(exception)),
             # invoke http call
             ops.map(lambda x: self._http(x)),
             # catch exception to fail batch response
diff --git a/tests/test_WriteApiBatching.py b/tests/test_WriteApiBatching.py
index ad453c2a..cb4c91a1 100644
--- a/tests/test_WriteApiBatching.py
+++ b/tests/test_WriteApiBatching.py
@@ -2,7 +2,12 @@
 
 from __future__ import absolute_import
 
+import http.server
+import os
+import subprocess
 import sys
+import textwrap
+import threading
 import time
 import unittest
 from collections import namedtuple
@@ -804,5 +809,88 @@ def test_flush_synchronous_is_noop(self):
         self._write_client.flush()
 
 
+class BatchingWriteInterpreterShutdownTest(unittest.TestCase):
+    """
+    Closing the batching WriteApi in an ``atexit`` handler writes the last batch.
+
+    Since Python 3.9 the thread pools used by the batching pipeline stop accepting work before ``atexit``
+    handlers run, so the scenario runs in a separate interpreter against a local HTTP server.
+    """
+
+    def setUp(self) -> None:
+        self.requests = []
+        requests = self.requests
+
+        class WriteHandler(http.server.BaseHTTPRequestHandler):
+            def do_POST(self):
+                requests.append(self.rfile.read(int(self.headers['Content-Length'])).decode('utf-8'))
+                self.send_response(204)
+                self.end_headers()
+
+            def log_message(self, format, *args):
+                pass
+
+        self.httpd = http.server.ThreadingHTTPServer(('127.0.0.1', 0), WriteHandler)
+        self.httpd_thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
+        self.httpd_thread.start()
+
+    def tearDown(self) -> None:
+        self.httpd.shutdown()
+        self.httpd.server_close()
+
+    def _run_script(self, script):
+        return subprocess.run([sys.executable, '-c', textwrap.dedent(script),
+                               'http://127.0.0.1:%d' % self.httpd.server_address[1]],
+                              cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=60)
+
+    def test_close_in_atexit(self):
+        result = self._run_script("""
+            import atexit
+            import sys
+
+            from influxdb_client import InfluxDBClient
+
+            client = InfluxDBClient(url=sys.argv[1], token="my-token", org="my-org")
+            write_api = client.write_api()
+
+            @atexit.register
+            def close():
+                print("closing")
+                write_api.close()
+                client.close()
+
+            write_api.write("my-bucket", record="h2o_feet,location=coyote_creek water_level=1 1")
+        """)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(["h2o_feet,location=coyote_creek water_level=1 1"], self.requests, result.stderr)
+        self.assertIn("closing", result.stdout)
+        self.assertNotIn("cannot schedule new futures", result.stderr)
+
+    def test_write_and_close_in_atexit(self):
+        result = self._run_script("""
+            import atexit
+            import sys
+
+            from influxdb_client import InfluxDBClient
+
+            client = InfluxDBClient(url=sys.argv[1], token="my-token", org="my-org")
+            write_api = client.write_api()
+
+            @atexit.register
+            def close():
+                print("closing")
+                write_api.write("my-bucket", record="h2o_feet,location=coyote_creek water_level=2 2")
+                write_api.close()
+                client.close()
+        """)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(["h2o_feet,location=coyote_creek water_level=2 2"], self.requests, result.stderr)
+        self.assertIn("closing", result.stdout)
+        self.assertNotIn("cannot schedule new futures", result.stderr)
+
+
 if __name__ == '__main__':
     unittest.main()