🔐 Sid Gifari File Manager Pro
v8.0.5 | 2026-08-08 16:46:08 | PHP 8.2.32
📂
/ (Root)
/
opt
/
alt
/
python38
/
lib
/
python3.8
/
site-packages
/
influxdb_client
/
client
/
__pycache__
📍 /opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/__pycache__
🔄 Refresh
✏️
Editing: write_api.cpython-38.pyc
Read Only
U v�iOb � @ s� d Z ddlZddlZddlZddlmZ ddlmZ ddlm Z ddl m Z ddl m Z ddl mZmZmZmZ ddlZdd lmZmZ dd lmZ dd lmZ dd lm Z dd l m Z m Z ddl m!Z! ddl"m#Z# ddl$m%Z%m&Z& ddl'm(Z( ddl)m*Z* e�+d�Z,e �r ddl-Z-ddl-m.Z. G dd� de �Z/G dd� de0�Z1e1e/j2d�Z3e1e/j4d�Z5G dd� de0�Z6G d d � d e0�Z7G d d � d e0�Z8G d d!� d!e0�Z9d"d#� Z:G d$d%� d%e �Z;dS )&zECollect and write time series data to InfluxDB Cloud or InfluxDB OSS.� N)� defaultdict)� timedelta)�Enum)�random)�sleep)�Union�Any�Iterable� NamedTuple)� operators� Observable)�ThreadPoolScheduler)�Subject)�WritePrecision)� _BaseWriteApi�_HAS_DATACLASS)�get_org_query_param)�DataframeSerializer)�Point�DEFAULT_WRITE_PRECISION)� WritesRetry)�_UTF_8_encodingz influxdb_client.client.write_api)� dataclassc @ s e Zd ZdZdZdZdZdS )� WriteTypez3Configuration which type of writes will client use.� � � N)�__name__� __module__� __qualname__�__doc__�batching� asynchronous� synchronous� r$ r$ �Q/opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/write_api.pyr # s r c @ sX e Zd ZdZejddddddddd ed d �f ed d �dd�Zdd� Zdd� Z dd� Z d S )� WriteOptionszWrite configuration.�� r i� � iH� i � r i� r �� max_workersN)� write_type�returnc C sF || _ || _|| _|| _|| _|| _|| _|| _| | _| | _ | | _ dS )aw Create write api configuration. :param write_type: methods of write (batching, asynchronous, synchronous) :param batch_size: the number of data point to collect in batch :param flush_interval: flush data at least in this interval (milliseconds) :param jitter_interval: this is primarily to avoid large write spikes for users running a large number of client instances ie, a jitter of 5s and flush duration 10s means flushes will happen every 10-15s (milliseconds) :param retry_interval: the time to wait before retry unsuccessful write (milliseconds) :param max_retries: the number of max retries when write fails, 0 means retry is disabled :param max_retry_delay: the maximum delay between each retry attempt in milliseconds :param max_retry_time: total timeout for all retry attempts in milliseconds, if 0 retry is disabled :param exponential_base: base for the exponential retry delay :parama max_close_wait: the maximum time to wait for writes to be flushed if close() is called :param write_scheduler: N) r+ � batch_size�flush_interval�jitter_interval�retry_interval� max_retries�max_retry_delay�max_retry_time�exponential_base�write_scheduler�max_close_wait) �selfr+ r- r. r/ r0 r1 r2 r3 r4 r6 r5 r$ r$ r% �__init__. s zWriteOptions.__init__c K s> t | j| jd | jd | jd | jd | j|�dd�dgd�S )a" Create a Retry strategy from write options. :key retry_callback: The callable ``callback`` to run after retryable error occurred. The callable must accept one argument: - `Exception`: an retryable error r' �retry_callbackN�POST)�totalr0 r/ r2 r3 r4 r9 �allowed_methods)r r1 r0 r/ r2 r3 r4 �get)r7 �kwargsr$ r$ r% �to_retry_strategyU s �z WriteOptions.to_retry_strategyc C s | j �� }|d= |S )�4Return a dict of attributes that you want to pickle.r5 ��__dict__�copy�r7 �stater$ r$ r% � __getstate__g s zWriteOptions.__getstate__c C s | j �|� tdd�| _dS )�'Set your object with the provided dict.r r) N)rB �updater r5 rD r$ r$ r% � __setstate__n s zWriteOptions.__setstate__) r r r r r r! r r8 r? rF rI r$ r$ r$ r% r& + s$ � � 'r& )r+ c @ s8 e Zd ZdZdd�dd�Zedd� �Zdd�dd �ZdS ) � PointSettingsz Settings to store default tags.N�r, c K s* t � | _|�� D ]\}}| �||� qdS )z� Create point settings for write api. :param default_tags: Default tags which will be added to each point written by api. N)�dict� defaultTags�items�add_default_tag)r7 Z default_tags�key�valr$ r$ r% r8 | s zPointSettings.__init__c C s" | � d�r tj�| dd� �S | S )Nz${env.� ���)� startswith�os�environr= )�valuer$ r$ r% � _get_value� s zPointSettings._get_valuec C s | � |�| j|< dS )z'Add new default tag with key and value.N)rX rM )r7 rP rW r$ r$ r% rO � s z PointSettings.add_default_tag)r r r r r8 � staticmethodrX rO r$ r$ r$ r% rJ y s rJ c @ sJ e Zd Zefdd�dd�Zed�dd�Zeed�dd �Z e d�d d �Z dS ) � _BatchItemKeyNrK c C s || _ || _|| _d S �N)�bucket�org� precision)r7 r\ r] r^ r$ r$ r% r8 � s z_BatchItemKey.__init__c C s t | j| j| jf�S r[ )�hashr\ r] r^ �r7 r$ r$ r% �__hash__� s z_BatchItemKey.__hash__)�or, c C s0 t || j�o.| j|jko.| j|jko.| j|jkS r[ )� isinstance� __class__r\ r] r^ )r7 rb r$ r$ r% �__eq__� s � � �z_BatchItemKey.__eq__c C s d� t| j�t| j�t| j��S )Nz4_BatchItemKey[bucket:'{}', org:'{}', precision:'{}'])�format�strr\ r] r^ r` r$ r$ r% �__str__� s �z_BatchItemKey.__str__) r r r r r8 �intra �object�boolre rg rh r$ r$ r$ r% rZ � s rZ c @ s@ e Zd Zd edd�dd�Zeeefd�dd�Zed�d d �ZdS ) � _BatchItemr N)rP r, c C s || _ || _|| _d S r[ �rP �data�size)r7 rP rn ro r$ r$ r% r8 � s z_BatchItem.__init__rK c C s | j j| j j| j jfS r[ )rP r\ r] r^ r` r$ r$ r% � to_key_tuple� s z_BatchItem.to_key_tuplec C s d� t| j�t| j��S )Nz _BatchItem[key:'{}', size: '{}'])rf rg rP ro r` r$ r$ r% rh � s �z_BatchItem.__str__)r )r r r rZ r8 rg rp rh r$ r$ r$ r% rl � s rl c @ s, e Zd Zdeed�dd�Zed�dd�ZdS ) �_BatchResponseN�rn � exceptionc C s || _ || _d S r[ rr )r7 rn rs r$ r$ r% r8 � s z_BatchResponse.__init__rK c C s d� | jrdndt| j��S )Nz!_BatchResponse[status:'{}', '{}']�failed�success)rf rs rg rn r` r$ r$ r% rh � s �z_BatchResponse.__str__)N)r r r rl � Exceptionr8 rg rh r$ r$ r$ r% rq � s rq c C s d� tdd� | ��S )N� c S s | j S r[ �rn �� batch_itemr$ r$ r% �<lambda>� � z _body_reduce.<locals>.<lambda>)�join�map)Z batch_itemsr$ r$ r% � _body_reduce� s r c s e Zd ZdZe� e� feedd�� fdd� Zddefeee ee d e e d e e d e e d eee d d e d f eed �d d�Zdd� Zdd� Zdd� Zdd� Zdd� Zefdd�Zed�d d �Zd d � Zee d �d!d"�Z d#d$� Z e d%�d&d'�Z e!d(d)� �Z"d*d+� Z#d,d-� Z$d.d/� Z%� Z&S )0�WriteApia� Implementation for '/api/v2/write' endpoint. Example: .. code-block:: python from influxdb_client import InfluxDBClient from influxdb_client.client.write_api import SYNCHRONOUS # Initialize SYNCHRONOUS instance of WriteApi with InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") as client: write_api = client.write_api(write_options=SYNCHRONOUS) N)� write_options�point_settingsr, c s� t � j||d� |� _|�dd�� _|�dd�� _|�dd�� _d� _� jjt j kr�t � � _ t d�� _� j �tj|jt|jd�� jd�t�d d � �t�d d � �tj� fd d �d �t�� ��� j� j� j�� _ n d� _ d� _ � jjt j kr�d}t � |t � dS )a� Initialize defaults. :param influxdb_client: with default settings (organization) :param write_options: write api configuration :param point_settings: settings to store default tags. :key success_callback: The callable ``callback`` to run after successfully writen a batch. The callable must accept two arguments: - `Tuple`: ``(bucket, organization, precision)`` - `str`: written data **[batching mode]** :key error_callback: The callable ``callback`` to run after unsuccessfully writen a batch. The callable must accept three arguments: - `Tuple`: ``(bucket, organization, precision)`` - `str`: written data - `Exception`: an occurred error **[batching mode]** :key retry_callback: The callable ``callback`` to run after retryable error occurred. The callable must accept three arguments: - `Tuple`: ``(bucket, organization, precision)`` - `str`: written data - `Exception`: an retryable error **[batching mode]** )�influxdb_clientr� �success_callbackN�error_callbackr9 r �Z milliseconds)�countZtimespan� schedulerc S s& | � t�dd� �t�dd� �t�� �S )Nc S s | j S r[ )rP ry r$ r$ r% r{ r| z5WriteApi.__init__.<locals>.<lambda>.<locals>.<lambda>c s � � t�� t�� fdd���S )Nc s t � jt| �t| �d�S )Nrm )rl rP r �len)�xs��groupr$ r% r{ r| zGWriteApi.__init__.<locals>.<lambda>.<locals>.<lambda>.<locals>.<lambda>)�pipe�opsZ to_iterabler~ r� r$ r� r% r{ s �)r� r� Zgroup_byr~ � merge_all)Zwindowr$ r$ r% r{ s �z#WriteApi.__init__.<locals>.<lambda>c S s | j dkS )Nr )ro ��batchr$ r$ r% r{ r| c s � j | � �� d�S )N�rn �delay)� _to_response� _jitter_delayr� r` r$ r% r{ r| )Zmapperz�The 'WriteType.asynchronous' is deprecated and will be removed in future major version. You can use native asynchronous version of the client: - https://influxdb-client.readthedocs.io/en/stable/usage.html#how-to-use-asyncio )!�superr8 �_write_optionsr= �_success_callback�_error_callback�_retry_callback�_window_schedulerr+ r r! r �_subjectr r� r� Zwindow_with_time_or_countr- r r. Zflat_map�filterr~ r� � subscribe�_on_next� _on_error� _on_complete� _disposabler" �warnings�warn�DeprecationWarning)r7 r� r� r� r> �message�rd r` r% r8 � s8 # � � �zWriteApi.__init__rg r rL �bytesr r )r\ r] �record�write_precisionr, c s� t ��jd����|� �jjtjkr:�j��||f|�S tt �}�j |||f|� �jjtj krfdnd� � ���fdd�}t t ||� � ��}� s�dS t|�dkr�|d S |S ) aQ Write time-series data into InfluxDB. :param str bucket: specifies the destination bucket for writes (required) :param str, Organization org: specifies the destination organization for writes; take the ID, Name or Organization. If not specified the default value from ``InfluxDBClient.org`` is used. :param WritePrecision write_precision: specifies the precision for the unix timestamps within the body line-protocol. The precision specified on a Point has precedes and is use for write. :param record: Point, Line Protocol, Dictionary, NamedTuple, Data Classes, Pandas DataFrame or RxPY Observable to write :key data_frame_measurement_name: name of measurement for writing Pandas DataFrame - ``DataFrame`` :key data_frame_tag_columns: list of DataFrame columns which are tags, rest columns will be fields - ``DataFrame`` :key data_frame_timestamp_column: name of DataFrame column which contains a timestamp. The column can be defined as a :class:`~str` value formatted as `2018-10-26`, `2018-10-26 12:00`, `2018-10-26 12:00:00-05:00` or other formats and types supported by `pandas.to_datetime <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.to_datetime.html#pandas.to_datetime>`_ - ``DataFrame`` :key data_frame_timestamp_timezone: name of the timezone which is used for timestamp column - ``DataFrame`` :key record_measurement_key: key of record with specified measurement - ``dictionary``, ``NamedTuple``, ``dataclass`` :key record_measurement_name: static measurement name - ``dictionary``, ``NamedTuple``, ``dataclass`` :key record_time_key: key of record with specified timestamp - ``dictionary``, ``NamedTuple``, ``dataclass`` :key record_tag_keys: list of record keys to use as a tag - ``dictionary``, ``NamedTuple``, ``dataclass`` :key record_field_keys: list of record keys to use as a field - ``dictionary``, ``NamedTuple``, ``dataclass`` Example: .. code-block:: python # Record as Line Protocol write_api.write("my-bucket", "my-org", "h2o_feet,location=us-west level=125i 1") # Record as Dictionary dictionary = { "measurement": "h2o_feet", "tags": {"location": "us-west"}, "fields": {"level": 125}, "time": 1 } write_api.write("my-bucket", "my-org", dictionary) # Record as Point from influxdb_client import Point point = Point("h2o_feet").tag("location", "us-west").field("level", 125).time(1) write_api.write("my-bucket", "my-org", point) DataFrame: If the ``data_frame_timestamp_column`` is not specified the index of `Pandas DataFrame <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.html>`_ is used as a ``timestamp`` for written data. The index can be `PeriodIndex <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.PeriodIndex.html#pandas.PeriodIndex>`_ or its must be transformable to ``datetime`` by `pandas.to_datetime <https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.to_datetime.html#pandas.to_datetime>`_. If you would like to transform a column to ``PeriodIndex``, you can use something like: .. code-block:: python import pandas as pd # DataFrame data_frame = ... # Set column as Index data_frame.set_index('column_name', inplace=True) # Transform index to PeriodIndex data_frame.index = pd.to_datetime(data_frame.index, unit='s') )r] �clientTFc s$ d� | d �}��� ��|| d �S )Nrw r r )r} � _post_write)�payloadZ final_string�� _async_reqr\ r] r7 r$ r% � write_payloady s z%WriteApi.write.<locals>.write_payloadNr r )r �_influxdb_clientZ_append_default_tagsr� r+ r r! �_write_batchingr �list� _serializer" r~ rN r� ) r7 r\ r] r� r� r> Zpayloadsr� �resultsr$ r� r% �write$ s$ H �� zWriteApi.writec C s dS )z Flush data.Nr$ r` r$ r$ r% �flush� s zWriteApi.flushc C s | � � dS )z)Flush data and dispose a batching buffer.N)�__del__r` r$ r$ r% �close� s zWriteApi.closec C s | S )u� Enter the runtime context related to this object. It will bind this method’s return value to the target(s) specified in the `as` clause of the statement. return: self instance r$ r` r$ r$ r% � __enter__� s zWriteApi.__enter__c C s | � � dS )zGExit the runtime context related to this object and close the WriteApi.N)r� )r7 �exc_type�exc_val�exc_tbr$ r$ r% �__exit__� s zWriteApi.__exit__c C s� | j rd| j �� | j �� d| _ | jjd }d}d}| jjsdt|� ||7 }||kr4t� d|� qdq4| j r�| j j j dd� d| _ | jr�d| _dS )zClose WriteApi.Nr' r g�������?zXReached max_close_wait (%s seconds) waiting for batches to finish writing. Force closingF)�wait) r� Z on_completed�disposer� r6 r� Z is_disposedr �logger�warningr� �executor�shutdown)r7 Z max_wait_timeZwaitedZ sleep_periodr$ r$ r% r� � s, �zWriteApi.__del__c s� t |t�r.t� ���}�j�t||d�� �n~t |t�rV�j� �|�t ��f�� �nVt |t �r~�j� �|� � |j f�� �n.t |t �r��j� �t j|fd�i����f�� n�dt|�jk�rt|�j��jjf��}t|j�D ] }�j� �|�|��f�� q�n�t|d��r*�j� �|�� �f�� n�t�rXt� |��rX�j� �t� |��f�� nTt |t ��r�|D ]} �j� �| �f�� �qhn&t |t ��r�|� � ����fdd�� d S )N)rP rn r� Z DataFrame�_asdictc s �j � �| �f��S r[ )r� )�it�r\ r> r] r^ r7 r$ r% r{ � r| z*WriteApi._write_batching.<locals>.<lambda>)!rc r� rZ r� Zon_nextrl rg r� �encoder r Zto_line_protocolr� rL � from_dict�typer r �_point_settingsr� r- �rangeZnumber_of_chunks� serialize�hasattrr� r � dataclasses� is_dataclass�asdictr r r� ) r7 r\ r] rn r^ r> �_key� serializerZ chunk_idx�itemr$ r� r% r� � sL �� ����� zWriteApi._write_batchingry c sr t �d� � �jr"� �fdd�}nd }�jj|d�}�jd� jj� jj� j � jj d|id� t �d� � t � d �S ) Nz(Write time series data into InfluxDB: %sc s �� � �� � j| �S r[ )r� rp rn )rs �rz r7 r$ r% �_retry_callback_delegate� s z0WriteApi._http.<locals>._retry_callback_delegate)r9 F�retries)� urlopen_kwzWrite request finished %srx ) r� �debugr� r� r? r� rP r\ r] rn r^ rq )r7 rz r� �retryr$ r� r% �_http� s � zWriteApi._httpc K s | j jf |||||dd�|��S )Nztext/plain; charset=utf-8)r] r\ �bodyr^ Z async_req� content_type)�_write_serviceZ post_write)r7 r� r\ r] r� r^ r> r$ r$ r% r� s ��zWriteApi._post_writer� c sL t �� ��t��jj�tj|�jjd�t��fdd��tj � fdd�d��S )N)Zduetimer� c s � � | �S r[ )r� )�xr` r$ r% r{ r| z'WriteApi._to_response.<locals>.<lambda>c s t �t| � d��S )N)rs rn )�rxZjustrq )rs �sourcerx r$ r% r{ r| )�handler) r� Zofr� r� Z subscribe_onr� r5 r� r~ Zcatch)r7 rn r� r$ )rn r7 r% r� s �zWriteApi._to_responsec C s t t� | jj d�S )Nr� )r r r� r/ r` r$ r$ r% r� s zWriteApi._jitter_delay)�responsec C s� |j rjt�d|j � | jr�z | �|j�� |jj|j � W q� tk rf } zt�d|� W 5 d }~X Y q�X n\t�d|� | jr�z| �|j�� |jj� W n. tk r� } zt�d|� W 5 d }~X Y nX d S )Nz8The batch item wasn't processed successfully because: %sz4The configured error callback threw an exception: %sz.The batch item: %s was processed successfully.z6The configured success callback threw an exception: %s) rs r� �errorr� rn rp rv r� r� )r7 r� �er$ r$ r% r� s zWriteApi._on_nextc C s t �d| � d S )Nz$unexpected error during batching: %s)r� r� )�exr$ r$ r% r� 0 s zWriteApi._on_errorc C s | j �� t�d� d S )Nz#the batching processor was disposed)r� r� r� �infor` r$ r$ r% r� 4 s zWriteApi._on_completec C s& | j �� }|d= |d= |d= |d= |S )r@ r� r� r� r� rA rD r$ r$ r% rF 8 s zWriteApi.__getstate__c C s2 | j �|� | j| j| j| j| j| j| jd� dS )rG )r� r� r9 N) rB rH r8 r� r� r� r� r� r� rD r$ r$ r% rI B s �zWriteApi.__setstate__)'r r r r r& rJ r8 r rg r r r rL r� r r r r r� r� r� r� r� r� r� rl r� r� r r� r� rq r� rY r� r� rF rI � __classcell__r$ r$ r� r% r� � sR ��O� �� � ` *� * r� )<r �loggingrU r� � collectionsr �datetimer �enumr r �timer �typingr r r r Z reactivexr� r r� r Zreactivex.schedulerr Zreactivex.subjectr r� r Z influxdb_client.client._baser r Z#influxdb_client.client.util.helpersr Z1influxdb_client.client.write.dataframe_serializerr Z"influxdb_client.client.write.pointr r Z"influxdb_client.client.write.retryr Zinfluxdb_client.restr � getLoggerr� r� r r rj r& r# Z SYNCHRONOUSr" Z ASYNCHRONOUSrJ rZ rl rq r r� r$ r$ r$ r% �<module> sD J
💾 Save Changes
❌ Cancel