U
vi @ s d Z ddlZddlmZmZ ddlmZ ddlmZm Z m
Z
ddl m Z m
Z
ddlmZmZ ddlmZ dd lmZ dd
lmZ ed Zerdd lmZ G d
d deZdS )zKCollect and async write time series data to InfluxDB Cloud or InfluxDB OSS. N)
ensure_futuregather) defaultdict)UnionIterable
NamedTuple)PointWritePrecision)
_BaseWriteApi_HAS_DATACLASS)get_org_query_param)DEFAULT_WRITE_PRECISION)
PointSettingsz&influxdb_client.client.write_api_async) dataclassc s e Zd ZdZe fedd fdd
Zddefeeeee d e
e d e e d e e d e
e d
d e d f eed d
dZ ZS )
WriteApiAsynca
Implementation for '/api/v2/write' endpoint.
Example:
.. code-block:: python
from influxdb_client_async import InfluxDBClientAsync
# Initialize async/await instance of Write API
async with InfluxDBClientAsync(url="http://localhost:8086", token="my-token", org="my-org") as client:
write_api = client.write_api()
N)point_settingsreturnc s t j||d dS )z
Initialize defaults.
:param influxdb_client: with default settings (organization)
:param point_settings: settings to store default tags.
)influxdb_clientr N)super__init__)selfr r __class__ W/opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/write_api_async.pyr " s zWriteApiAsync.__init__strr dictbytesr r )bucketorgrecordwrite_precisionr c
s t || jd}| | tt}| j|||fddi| g }| D ]2\}} |t| j j
||d | |dddd qFt |ddiI d H }
|
D ]} t
| tr| qdd
d |
D kS ) a
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 ``InfluxDBClientAsync.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
: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``
:return: ``True`` for successfully accepted data, otherwise raise an exception
Example:
.. code-block:: python
# Record as Line Protocol
await 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
}
await 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)
await 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 clientZprecision_from_pointT
Fztext/plain; charset=utf-8)r r body precisionZ async_reqZ_return_http_data_only content_typereturn_exceptionsNc S s g | ]}|d dkqS ) ) r ).0rer r r
s z'WriteApiAsync.write..)r Z_influxdb_clientZ_append_default_tagsr list
_serializeitemsappendr Z_write_serviceZpost_write_asyncjoinr
isinstance Exception) r r r r r! kwargsZpayloadsfuturesZpayload_precisionZ payload_lineresultsresultr r r write+ s( G
zWriteApiAsync.write)__name__
__module__ __qualname____doc__r r r
r r r r r r r r boolr9
__classcell__r r r r r s r ) r= loggingasyncior r collectionsr typingr r r r r r Z influxdb_client.client._baser
r Z#influxdb_client.client.util.helpersr Z"influxdb_client.client.write.pointr
Z influxdb_client.client.write_apir getLoggerlogger dataclassesr r r r r r s