refactor: multiprocess helper class#226
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #226 +/- ##
==========================================
- Coverage 87.21% 86.65% -0.56%
==========================================
Files 28 28
Lines 2072 2068 -4
==========================================
- Hits 1807 1792 -15
- Misses 265 276 +11 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Pull request overview
This PR refactors the MultiprocessingWriter helper to manage its own multiprocessing context/process (instead of subclassing multiprocessing.Process), updates the integration test accordingly, and records the change in the changelog.
Changes:
- Refactor
MultiprocessingWriterto usemultiprocessing.get_context(start_method)and aJoinableQueue, with a configurable start method (defaultspawn). - Update the integration test to exercise the refactored multiprocessing writer.
- Add coverage configuration for multiprocessing and document the refactor in
CHANGELOG.md.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
influxdb_client_3/write_client/client/util/multiprocessing_helper.py |
Refactors multiprocessing writer lifecycle/process creation and switches to direct WriteApi usage in the worker process. |
tests/test_influxdb_client_3_integration.py |
Updates integration test setup and expectations for the refactored multiprocessing helper. |
pyproject.toml |
Adds coverage configuration for multiprocessing runs. |
CHANGELOG.md |
Adds an entry describing the MultiprocessingWriter refactor. |
Comments suppressed due to low confidence (1)
influxdb_client_3/write_client/client/util/multiprocessing_helper.py:147
- Using
assertfor runtime state validation in library code is unsafe because asserts can be stripped with Python optimizations (-O), which would allow writes after disposal / before start. Prefer explicit checks that raise a real exception.
assert self.__disposed__ is False, 'Cannot write data: the writer is closed.'
assert self.__started__ is True, 'Cannot write data: the writer is not started.'
self.queue_.put(kwargs)
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| Initialize defaults. | ||
|
|
||
| For more information how to initialize the writer see the examples above. | ||
| For more information on how to initialize the writer, see the examples above. | ||
|
|
||
| :param kwargs: arguments are passed into ``__init__`` function of ``InfluxDBClient`` and ``write_api``. | ||
| :param kwargs: arguments are passed into the `` _ _init__`` function of ``InfluxDBClient`` and ``write_api``. |
| For more information on how to pass arguments, see the examples above. | ||
|
|
||
| :param kwargs: arguments are passed into ``write`` function of ``WriteApi`` | ||
| :param kwargs: arguments are passed into the `` write `` function of ``WriteApi`` | ||
| :return: None |
| self.write_api = WriteApi( | ||
| bucket=self.kwargs.get('database'), | ||
| org=self.kwargs.get('org'), | ||
| default_header=self.kwargs.get('default_header'), | ||
| rest_client=self.kwargs.get('rest_client'), |
| if self.__started__: | ||
| self.queue_.put(_PoisonPill()) | ||
| self.queue_.join() | ||
| self.join() | ||
| self.process.join() | ||
| self.queue_ = None | ||
| self.__started__ = False |
| 1. [#222](https://github.com/InfluxCommunity/influxdb3-python/pull/222): Refactor `MultiprocessingWriter` class. | ||
| - New Process will be created by using `DefaultContext.Process(target)` to prevent erratic crashes, handle cross-platform code behavior safely, and coordinate complex resource sharing. | ||
| - Users can now choose one of the start methods `fork`, `spawn` or `forkserver` when creating new Process. The default will be `spawn`. |
81d5997 to
0a46307
Compare
Closes #
Proposed Changes
Briefly describe your proposed changes:
Checklist