Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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]

Expand Down
9 changes: 9 additions & 0 deletions influxdb_client/client/write_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
88 changes: 88 additions & 0 deletions tests/test_WriteApiBatching.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Loading