U
vi @ s d Z ddlZddlZddlmZmZ ddlmZ edZ e
e
e
fe
dddZ e
e
e
fe
ed d
d Z e
e
e
fe
ed d d
Z
G dd dZG dd dejZdS )z
Helpers classes to make easier use the client in multiprocessing environment.
For more information how the multiprocessing works see Python's
`reference docs `_.
N)InfluxDBClient WriteOptions)
InfluxDBErrorz2influxdb_client.client.util.multiprocessing_helperconfdatac C s t d| d| dS )zSuccessfully writen batch.zWritten batch: , data: Nloggerdebugr r c/opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/util/multiprocessing_helper.py_success_callback s r r r exceptionc C s t d| d| d| dS )z Unsuccessfully writen batch.zCannot write batch: r z due: Nr r r r r
_error_callback s r c C s t d| d| d| dS )zRetryable error.z"Retryable error occurs for batch: r z retry: Nr r r r r
_retry_callback s r c @ s e Zd ZdZdS ) _PoisonPillz To notify process to terminate.N)__name__
__module__ __qualname____doc__r r r r
r s r c sx e Zd ZdZdZdZddddZddddZd d
Zdd fd d Z ddd
dZ
dd Z dd Z dd Z
ZS )MultiprocessingWriteraA
The Helper class to write data into InfluxDB in independent OS process.
Example:
.. code-block:: python
from influxdb_client import WriteOptions
from influxdb_client.client.util.multiprocessing_helper import MultiprocessingWriter
def main():
writer = MultiprocessingWriter(url="http://localhost:8086", token="my-token", org="my-org",
write_options=WriteOptions(batch_size=100))
writer.start()
for x in range(1, 1000):
writer.write(bucket="my-bucket", record=f"mem,tag=a value={x}i {x}")
writer.__del__()
if __name__ == '__main__':
main()
How to use with context_manager:
.. code-block:: python
from influxdb_client import WriteOptions
from influxdb_client.client.util.multiprocessing_helper import MultiprocessingWriter
def main():
with MultiprocessingWriter(url="http://localhost:8086", token="my-token", org="my-org",
write_options=WriteOptions(batch_size=100)) as writer:
for x in range(1, 1000):
writer.write(bucket="my-bucket", record=f"mem,tag=a value={x}i {x}")
if __name__ == '__main__':
main()
How to handle batch events:
.. code-block:: python
from influxdb_client import WriteOptions
from influxdb_client.client.exceptions import InfluxDBError
from influxdb_client.client.util.multiprocessing_helper import MultiprocessingWriter
class BatchingCallback(object):
def success(self, conf: (str, str, str), data: str):
print(f"Written batch: {conf}, data: {data}")
def error(self, conf: (str, str, str), data: str, exception: InfluxDBError):
print(f"Cannot write batch: {conf}, data: {data} due: {exception}")
def retry(self, conf: (str, str, str), data: str, exception: InfluxDBError):
print(f"Retryable error occurs for batch: {conf}, data: {data} retry: {exception}")
def main():
callback = BatchingCallback()
with MultiprocessingWriter(url="http://localhost:8086", token="my-token", org="my-org",
success_callback=callback.success,
error_callback=callback.error,
retry_callback=callback.retry) as writer:
for x in range(1, 1000):
writer.write(bucket="my-bucket", record=f"mem,tag=a value={x}i {x}")
if __name__ == '__main__':
main()
FN)returnc K s0 t j| || _d| _d| _t | _dS )z
Initialize defaults.
For more information how to initialize the writer see the examples above.
:param kwargs: arguments are passed into ``__init__`` function of ``InfluxDBClient`` and ``write_api``.
N) multiprocessingProcess__init__kwargsclient write_apiManagerQueuequeue_selfr r r r
r y s
z MultiprocessingWriter.__init__c K s4 | j dkstd| jdks$td| j| dS )z
Append time-series data into underlying queue.
For more information how to pass arguments see the examples above.
:param kwargs: arguments are passed into ``write`` function of ``WriteApi``
:return: None
Fz(Cannot write data: the writer is closed.Tz-Cannot write data: the writer is not started.N) __disposed__AssertionError __started__r" putr# r r r
write s zMultiprocessingWriter.writec C s t f | j| _| jj| jdt | jdt| jdt| jdtd| _| j }t
|t krx| | j
q| jjf | | j
qNdS )zIInitialize ``InfluxDBClient`` and waits for data to writes into InfluxDB.
write_optionssuccess_callbackerror_callbackretry_callback)r* r+ r, r- N)r r r r getr r r r r" typer terminate task_doner) )r$ Z next_recordr r r
run s
zMultiprocessingWriter.runc s t d| _dS )z9Start independent process for writing data into InfluxDB.TN)superstartr' r$ __class__r r
r4 s
zMultiprocessingWriter.startc C sD | j r td | j d| _ | jr@| j d| _td dS )z
Cleanup resources in independent process.
This function **cannot be used** to terminate the ``MultiprocessingWriter``.
If you want to finish your writes please call: ``__del__``.
zflushing data...Nclosed)r r
info__del__r r5 r r r
r0 s
z MultiprocessingWriter.terminatec C s | | S )z1Enter the runtime context related to this object.)r4 r5 r r r
__enter__ s z MultiprocessingWriter.__enter__c C s | dS )z0Exit the runtime context related to this object.N)r: )r$ exc_type exc_value tracebackr r r
__exit__ s z MultiprocessingWriter.__exit__c C s< | j r,| jt | j | d| _d| _ d| _dS )z!Dispose the client and write_api.NFT)r' r" r( r joinr% r5 r r r
r: s
z MultiprocessingWriter.__del__)r r r r r' r% r r) r2 r4 r0 r; r? r:
__classcell__r r r6 r
r % s P
r )r loggingr Zinfluxdb_clientr r Z!influxdb_client.client.exceptionsr getLoggerr
strr r r r r r r r r r
s