refactor: multiprocessing helper

[NguyenHoangSon96]

Closes #

Proposed Changes

• Refactor MultiprocessingWriter class:
  - WriteApi will be created and used directly in the class for writing.
  - Tests are added for writing with multiprocessing.
  - Use DefaultContext.Process(target) to create a new process.
  - Users can now choose one of the start methods fork, spawn or forkserver when creating a new Process. The default will be spawn.
  - Implement a time-to live mechanism for child processes. I just set timeout for JoinableQueue.get() so It will wait for a number of times, default is 300 seconds, and users can pass in their own value to override that by process_ttl parameter.
  - Users can pass in an on_shutdown() function which gets called when the queue has been shutdown.

Checklist

• CHANGELOG.md updated
• Rebased/mergeable
• A test has been added if appropriate
• Tests pass
• Commit messages are conventional
• Sign CLA (if not already signed)

[codecov]

Codecov Report

❌ Patch coverage is 97.67442% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 86.88%. Comparing base (b24d6c1) to head (8bda24e).

Files with missing lines	Patch %	Lines
...write_client/client/util/multiprocessing_helper.py	97.67%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #222      +/-   ##
==========================================
- Coverage   87.21%   86.88%   -0.33%     
==========================================
  Files          28       28              
  Lines        2072     2082      +10     
==========================================
+ Hits         1807     1809       +2     
- Misses        265      273       +8     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

This PR refactors the write stack by removing the generated OpenAPI-style service/client layer and moving write request construction and HTTP dispatch directly into WriteApi, while also updating tests and documentation to reflect the new behavior and configuration surface.

Changes:

• Refactors WriteApi to perform request preparation + HTTP calls directly via a new low-level RestClient (removing ApiClient/WriteService/Configuration).
• Updates InfluxDBClient3 initialization and tests to use the new header/rest client approach and to validate required constructor keys.
• Refactors and adds integration coverage for MultiprocessingWriter start-method behavior and direct WriteApi usage.

Reviewed changes

Copilot reviewed 25 out of 25 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
CHANGELOG.md	Documents breaking changes and new construction patterns (example needs a small syntax fix).
influxdb_client_3/init.py	Reworks InfluxDBClient3 construction to create RestClient + WriteApi directly and enforce required keys.
influxdb_client_3/write_client/init.py	Removes deprecated exports and re-exports updated WriteApi/types.
influxdb_client_3/write_client/_sync/api_client.py	Removes legacy generated ApiClient.
influxdb_client_3/write_client/_sync/rest.py	Removes legacy generated REST client implementation.
influxdb_client_3/write_client/_sync/rest_client.py	Adds new RestClient for low-level HTTP handling + logging/redaction.
influxdb_client_3/write_client/client/init.py	Removes re-export of removed WriteService.
influxdb_client_3/write_client/client/_base.py	Removes legacy base client/write-api helpers tied to generated stack.
influxdb_client_3/write_client/client/influxdb_client.py	Removes legacy v2 InfluxDBClient.
influxdb_client_3/write_client/client/logging_handler.py	Removes InfluxLoggingHandler implementation.
influxdb_client_3/write_client/client/util/helpers.py	Updates exception import path.
influxdb_client_3/write_client/client/util/multiprocessing_helper.py	Refactors MultiprocessingWriter to create WriteApi directly and set default start method.
influxdb_client_3/write_client/client/write/init.py	Removes legacy re-export of removed WriteService.
influxdb_client_3/write_client/client/write_api.py	Major refactor: WriteApi becomes standalone, builds write requests and calls RestClient directly, retains batching pipeline.
influxdb_client_3/write_client/configuration.py	Removes legacy generated configuration object.
influxdb_client_3/write_client/service/init.py	Removes legacy service package init exporting WriteService.
influxdb_client_3/write_client/service/_base_service.py	Removes legacy service base class.
influxdb_client_3/write_client/service/write_service.py	Removes legacy generated WriteService.
influxdb_client_3/write_client/write_exceptions.py	Removes _BaseRESTClient logger helper (moved to RestClient).
tests/test_influxdb_client_3.py	Updates unit tests for new client/write header/rest behavior and required params.
tests/test_influxdb_client_3_integration.py	Adds integration tests for direct WriteApi usage, multiprocessing helper, and async call path.
tests/test_polars.py	Updates write tests to mock WriteApi call sites instead of WriteService.
tests/test_query.py	Fixes typo in client init parameter (database).
tests/test_write_api.py	Refactors tests from generated service/client usage to InfluxDBClient3/WriteApi usage.
tests/test_write_local_server.py	Updates exception import path.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 25 out of 25 changed files in this pull request and generated 3 comments.

[NguyenHoangSon96]

@bednar @karel-rehor @alespour
Hi guys,
I made some quite big changes to our MultiprocessingWriter class.
Please take a look. I'm not very familiar with multiprocessing in Python.

[karel-rehor]

This looks good so far.

I especially like the use of get_context() for each instance. It's cleaner, more flexible and more stable than what I suggested last week.

A couple of notes.

1. need to update CHANGELOG.md

2. The codecov report shows that important parts of the run and terminate methods are not covered by tests.

I think this is partially a false report. I suspect codecov is not tracking code executed in newly spawned processes. When running tests in the debugger, I see the processes enter this area of code. It's probably necessary to configure codecov to follow multiprocess executions.

e.g. in pyproject.toml (+AI+ suggestion 1)

[tool.coverage.run]
concurrency = ["multiprocessing"]
parallel = true
source = ["your_package_name"]

or perhaps in .coveragerc (+AI+ suggestion 2)

[run]
concurrency = multiprocessing
parallel = true
source = your_package_name

It was suggested in a PR for influxdb-client-php that CircleCI has a codecov orb that simplifies working with codecoverage reports. It might help resolve this issue.

[karel-rehor]

I'm not comfortable with the implication that end users will be forced to work directly with the transport level RestClient class. The rest_client kwarg should be optional. A default instance should be generated using other arguments passed to the MultiprocessingWriter constructor.

[karel-rehor]

At this point it looks like the kwarg rest_client is required for the MultiprocessingWriter to work. It MUST be supplied to the constructor as a part of self.kwargs. However, RestClient is the lowest level transport API object used in the communications stack, and in #217 in the CHANGELOG.md it was mentioned that end users should rarely have to use this directly.

Since the constructor for RestClient essentially uses the host and default_header properties also passed to the MultiprocessingWriter constructor, couldn't self.rest_client simply be instantiated in __init__ or generated by default for the rest_client argument passed in run to the WriteApi constructor? In this way the end user will not be required to create his or her own RestClient instance in order to use the MultiprocessingWriter class. In generally, I think the RestClient API should remain hidden or internal unless users have a good reason to create their own.

BTW I see that when running test_multiprocessing_helper with the rest_client argument commented out, the test appears to run without ending. I let it run for (0:06:43) before sending a keyboard interrupt.

[karel-rehor]

After running the test test_multiprocessing_helper with the argument rest_client commented out, and seeing that the process can run almost indefinitely without communicating with the server, it occurs to me that the process also needs a TTL value, which might be reset after each successful write.

It is possible that process behind the multiprocessor can get stuck for a number of reasons, and it might be best to force MultiprocessingWriter to shut itself down in such a case.

For example if TTL is 5 minutes and the process writes nothing for 5 minutes, then a poison pill is automatically generated to terminate the process. In favorable conditions after a successful write is made, the counter gets reset and another 5 minute window begins. A TTL of 0 would mean that this check is switched off.

After a bit of reflection, in addition to a TTL value it might also be nice to add a user defined function (e.g. on_shutdown() passed in constructor) to be called whenever the process shuts itself down. In that way the user might be able to log important information or send notifications.

[karel-rehor]

Does this imply that users will also be required to instantiate their own RestClient in order to use MultiprocessingWriter? See comment in multiprocessing-helper.py

[NguyenHoangSon96]

Hi @karel-rehor

CHANGES:

• Move WriteApi initiation steps outside of the child process so it can fail fast inside the parent process if there are any issues.
• Implement a time-to live mechanism for child processes. I just set timeout for JoinableQueue.get() so It will wait for a number of times, default is 300 seconds, and users can pass in their own value to override that by process_ttl param.
• Users can pass in an on_shutdown() function which gets called when the queue has been shutdown.
• Users don't have to manually provide RestClient anymore.
• Tests added for those changes.

TODO:

• Update CHANGELOG.md and this PR description section.

Please click "Resolve conversation" after all issues have been fixed.

[karel-rehor]

I still have a couple of minor requests, regarding tests and CHANGELOG.md.

[karel-rehor]

Maybe I'm paranoid, but I might add a preliminary assert, just to ensure that the original state of number is as expected.

e.g.

            self.assertEqual(number.value, 0)

Then we can be doubly assured that the callback changed the state of the variable. Not a blocker though.

[karel-rehor]

Instead of skipping this test, can the timeout argument in localClient.query() be shortened to say 10 microseconds 0.00001 or even a microsecond 10^-6. I see this still works for me locally adding finer and finer values. Ideally the value for this test could be any value in lim x -> 0, of course without reaching 0, that wins the race against the query response.

[karel-rehor]

Need to mention constructor arguments ttl and on_shutdown.