🔐 Sid Gifari File Manager Pro
v8.0.5 | 2026-08-09 09:45:53 | PHP 8.2.33
📂
/ (Root)
/
opt
/
alt
/
python38
/
lib
/
python3.8
/
site-packages
/
influxdb_client
/
client
/
util
/
__pycache__
📍 /opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/util/__pycache__
🔄 Refresh
✏️
Editing: multiprocessing_helper.cpython-38.pyc
Read Only
U v�i� � @ s� d Z ddlZddlZddlmZmZ ddlmZ e�d�Z e e e fe d�dd�Z e e e fe ed �d d �Z e e e fe ed �d d �Z G dd� d�ZG dd� dej�ZdS )z� Helpers classes to make easier use the client in multiprocessing environment. For more information how the multiprocessing works see Python's `reference docs <https://docs.python.org/3/library/multiprocessing.html>`_. � N)�InfluxDBClient� WriteOptions)� InfluxDBErrorz2influxdb_client.client.util.multiprocessing_helper��conf�datac C s t �d| � d|� �� dS )zSuccessfully writen batch.zWritten batch: �, data: N��logger�debugr � 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 ZdZdZdZdd�dd�Zdd�dd�Zd d � Zdd�� fd d � Z dd�d d�Z 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) �multiprocessing�Process�__init__�kwargs�client� write_api�Manager�Queue�queue_��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| j�dt� �| j�dt�| j�dt�| j�dt�d�| _| j �� }t |�t krx| � � | j � � q�| jjf |� | j � � qNdS )zIInitialize ``InfluxDBClient`` and waits for data to writes into InfluxDB.� write_options�success_callback�error_callback�retry_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)�super�startr' �r$ �� __class__r r r4 � s zMultiprocessingWriter.startc C sD | j r t�d� | j �� d| _ | jr@| j�� d| _t�d� 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...N�closed)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,| j�t� � | 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 �<module> s
💾 Save Changes
❌ Cancel