U
viOb @ 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+dZ,e r ddl-Z-ddl-m.Z. G dd de Z/G dd de0Z1e1e/j2dZ3e1e/j4dZ5G dd de0Z6G d d d e0Z7G d d d e0Z8G d d! d!e0Z9d"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)UnionAnyIterable
NamedTuple) operators
Observable)ThreadPoolScheduler)Subject)WritePrecision)
_BaseWriteApi_HAS_DATACLASS)get_org_query_param)DataframeSerializer)PointDEFAULT_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
ddZdd Zdd Z dd Z
d S ) WriteOptionszWrite configuration. r i iH i r i r max_workersN)
write_typereturnc 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_sizeflush_intervaljitter_intervalretry_interval max_retriesmax_retry_delaymax_retry_timeexponential_basewrite_schedulermax_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|dddgdS )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_callbackNPOST)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__copyr7 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 ZdZddddZedd Zdddd ZdS )
PointSettingsz Settings to store default tags.Nr, 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 defaultTagsitemsadd_default_tag)r7 Z default_tagskeyvalr$ r$ r% r8 | s zPointSettings.__init__c C s" | dr tj| dd S | S )Nz${env. )
startswithosenvironr= )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 ZefddddZedddZeeddd Z e
dd
d Z dS )
_BatchItemKeyNrK c C s || _ || _|| _d S N)bucketorg precision)r7 r\ r] r^ r$ r$ r% r8 s z_BatchItemKey.__init__c C s t | j| j| jfS r[ )hashr\ r] r^ r7 r$ r$ r% __hash__ s z_BatchItemKey.__hash__)or, c C s0 t || jo.| 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| jt| jt| jS )Nz4_BatchItemKey[bucket:'{}', org:'{}', precision:'{}'])formatstrr\ r] r^ r` r$ r$ r% __str__ s
z_BatchItemKey.__str__) r r r r r8 intra objectboolre rg rh r$ r$ r$ r% rZ s rZ c @ s@ e Zd Zd eddddZeeefdddZedd d
ZdS )
_BatchItemr N)rP r, c C s || _ || _|| _d S r[ rP datasize)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| jt| jS )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 ZdeedddZedddZdS ) _BatchResponseNrn exceptionc C s || _ || _d S r[ rr )r7 rn rs r$ r$ r% r8 s z_BatchResponse.__init__rK c C s d | jrdndt| jS )Nz!_BatchResponse[status:'{}', '{}']failedsuccess)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% z _body_reduce..)joinmap)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
dZdd Zdd Zdd Zdd Zdd ZefddZedd 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 )0WriteApia
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_optionspoint_settingsr, c s t j||d | _|dd _|dd _|dd _d _ jjt j
krt _ t
d _ j tj|jt|jd jdtd d
td d
tj fd d
d
t j j j _ n d _ d _ jjt j krd}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_callbackNerror_callbackr9 r Z milliseconds)countZtimespan schedulerc S s& | tdd tdd t S )Nc S s | j S r[ )rP ry r$ r$ r% r{ r| z5WriteApi.__init__....c s t t fddS )Nc s t jt| t| dS )Nrm )rl rP r len)xsgroupr$ r% r{ r| zGWriteApi.__init__......)pipeopsZ to_iterabler~ r r$ r r% r{ s )r r Zgroup_byr~ merge_all)Zwindowr$ r$ r% r{
s z#WriteApi.__init__..c S s
| j dkS )Nr )ro batchr$ r$ r% r{ r| c s j | dS )Nrn delay) _to_response
_jitter_delayr r` r$ r% r{ r| )ZmapperzThe '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_mapfilterr~ r subscribe_on_next _on_error _on_complete _disposabler" warningswarnDeprecationWarning)r7 r r r r> messagerd r` r% r8 s8 #
zWriteApi.__init__rg r rL bytesr
r )r\ r] recordwrite_precisionr, c s t jd| jjtjkr:j||f|S tt }j
|||f| jjtj krfdnd fdd}t t ||
} sdS 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 `_ - ``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 `_
is used as a ``timestamp`` for written data. The index can be `PeriodIndex `_
or its must be transformable to ``datetime`` by
`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..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_typeexc_valexc_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| _
| jrd| _dS )zClose WriteApi.Nr' r g?zXReached max_close_wait (%s seconds) waiting for batches to finish writing. Force closingF)wait)
r Z on_completeddisposer r6 r Z is_disposedr loggerwarningr executorshutdown)r7 Z
max_wait_timeZwaitedZ sleep_periodr$ r$ r% r s,
zWriteApi.__del__c
s t |tr.t }jt||d n~t |trVj |t f nVt |t
r~j | |j f n.t |t
rj t
j|fdif ndt|jkrt|jjjf}t|jD ] }j ||f qnt|dr*j | f ntrXt |rXj t |f nTt |t r|D ]} j | f qhn&t |t r| fdd d S )N)rP rn r Z DataFrame_asdictc s j | fS r[ )r )itr\ r> r] r^ r7 r$ r% r{ r| z*WriteApi._write_batching..)!rc r rZ r Zon_nextrl rg r encoder r Zto_line_protocolr rL from_dicttyper r _point_settingsr r- rangeZnumber_of_chunks serializehasattrr r dataclasses is_dataclassasdictr r r )
r7 r\ r] rn r^ r> _key
serializerZ chunk_idxitemr$ 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.._retry_callback_delegate)r9 Fretries)
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 tjjtj|jjdtfddtj fdddS )N)Zduetimer c s
| S r[ )r )xr` r$ r% r{ r| z'WriteApi._to_response..c s t t| dS )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 dS )Nr )r r r r/ r` r$ r$ r% r s zWriteApi._jitter_delay)responsec
C s |j rjtd|j | jrz | |j |jj|j W q tk
rf } ztd| W 5 d }~X Y qX n\td| | jrz| |j |jj W n. tk
r } ztd| 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 td 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 ) sD
J