U
vi?0 @ sl d Z ddlmZmZ ddlmZ ddlmZmZ ddl m
Z
ddl m Z m
Z
ddlmZ G d d
d
eZd S ) u
Querying InfluxDB by FluxLang.
Flux is InfluxData’s functional data scripting language designed for querying, analyzing, and acting on data.
)ListAsyncGenerator)
_BaseQueryApi)
FluxRecord TableList) QueryOptions)_UTF_8_encoding ApiException )RESTResponseAsyncc s e Zd ZdZe f fdd ZdeeedddZ deee
d dd d
Z dee e ee
d d
dZdee e ee
d ddZdejdfeedddZdd Z ZS )
QueryApiAsyncz9Asynchronous implementation for '/api/v2/query' endpoint.c s t j||d dS )z[
Initialize query client.
:param influxdb_client: influxdb client
)influxdb_client
query_optionsN)super__init__)selfr
r __class__ W/opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/query_api_async.pyr s zQueryApiAsync.__init__N)queryparamsreturnc sB | |}| j|| || j|dI dH }| j|| dI dH S )a
Execute asynchronous Flux query and return result as a :class:`~influxdb_client.client.flux_table.FluxTable` list.
:param query: the Flux query
:param str, Organization org: specifies the organization for executing the query;
Take the ``ID``, ``Name`` or ``Organization``.
If not specified the default value from ``InfluxDBClientAsync.org`` is used.
:param params: bind parameters
:return: :class:`~influxdb_client.client.flux_table.FluxTable` list wrapped into
:class:`~influxdb_client.client.flux_table.TableList`
:rtype: TableList
Serialization the query results to flattened list of values via :func:`~influxdb_client.client.flux_table.TableList.to_values`:
.. code-block:: python
from influxdb_client import InfluxDBClient
async with InfluxDBClientAsync(url="http://localhost:8086", token="my-token", org="my-org") as client:
# Query: using Table structure
tables = await client.query_api().query('from(bucket:"my-bucket") |> range(start: -10m)')
# Serialize to values
output = tables.to_values(columns=['location', '_time', '_value'])
print(output)
.. code-block:: python
[
['New York', datetime.datetime(2022, 6, 7, 11, 3, 22, 917593, tzinfo=tzutc()), 24.3],
['Prague', datetime.datetime(2022, 6, 7, 11, 3, 22, 917593, tzinfo=tzutc()), 25.3],
...
]
Serialization the query results to JSON via :func:`~influxdb_client.client.flux_table.TableList.to_json`:
.. code-block:: python
from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
async with InfluxDBClientAsync(url="http://localhost:8086", token="my-token", org="my-org") as client:
# Query: using Table structure
tables = await client.query_api().query('from(bucket:"my-bucket") |> range(start: -10m)')
# Serialize to JSON
output = tables.to_json(indent=5)
print(output)
.. code-block:: javascript
[
{
"_measurement": "mem",
"_start": "2021-06-23T06:50:11.897825+00:00",
"_stop": "2021-06-25T06:50:11.897825+00:00",
"_time": "2020-02-27T16:20:00.897825+00:00",
"region": "north",
"_field": "usage",
"_value": 15
},
{
"_measurement": "mem",
"_start": "2021-06-23T06:50:11.897825+00:00",
"_stop": "2021-06-25T06:50:11.897825+00:00",
"_time": "2020-02-27T16:20:01.897825+00:00",
"region": "west",
"_field": "usage",
"_value": 10
},
...
]
orgr Nr )
_org_param _post_query
_create_querydefault_dialectZ_to_tables_async_get_query_optionsr r r r responser r r r s J
zQueryApiAsync.query)r Nc sB | |}| j|| || j|dI dH }| j|| dI dH S )a
Execute asynchronous Flux query and return stream of :class:`~influxdb_client.client.flux_table.FluxRecord` as an AsyncGenerator[:class:`~influxdb_client.client.flux_table.FluxRecord`].
:param query: the Flux query
:param str, Organization org: specifies the organization for executing the query;
Take the ``ID``, ``Name`` or ``Organization``.
If not specified the default value from ``InfluxDBClientAsync.org`` is used.
:param params: bind parameters
:return: AsyncGenerator[:class:`~influxdb_client.client.flux_table.FluxRecord`]
r Nr )r r r r Z _to_flux_record_stream_asyncr r! r r r query_streamj s
zQueryApiAsync.query_streamF)r data_frame_indexr use_extension_dtypesc sF | j |||||dI dH }g }|2 z3 dH W }|| q"6 | |S )a
Execute asynchronous Flux query and return :class:`~pandas.core.frame.DataFrame`.
.. note:: If the ``query`` returns tables with differing schemas than the client generates a :class:`~DataFrame` for each of them.
:param query: the Flux query
:param str, Organization org: specifies the organization for executing the query;
Take the ``ID``, ``Name`` or ``Organization``.
If not specified the default value from ``InfluxDBClientAsync.org`` is used.
:param data_frame_index: the list of columns that are used as DataFrame index
:param params: bind parameters
:param use_extension_dtypes: set to ``True`` to use panda's extension data types.
Useful for queries with ``pivot`` function.
When data has missing values, column data type may change (to ``object`` or ``float64``).
Nullable extension types (``Int64``, ``Float64``, ``boolean``) support ``panda.NA`` value.
For more info, see https://pandas.pydata.org/docs/user_guide/missing_data.html.
:return: :class:`~DataFrame` or :class:`~List[DataFrame]`
.. warning:: For the optimal processing of the query results use the ``pivot() function`` which align results as a table.
.. code-block:: text
from(bucket:"my-bucket")
|> range(start: -5m, stop: now())
|> filter(fn: (r) => r._measurement == "mem")
|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
For more info see:
- https://docs.influxdata.com/resources/videos/pivots-in-flux/
- https://docs.influxdata.com/flux/latest/stdlib/universe/pivot/
- https://docs.influxdata.com/flux/latest/stdlib/influxdata/influxdb/schema/fieldsascols/
)r r$ r r% N)query_data_frame_streamappendZ_to_data_frames) r r r r$ r r% Z
_generatorZ
dataframesZ dataframer r r query_data_frame{ s "
z QueryApiAsync.query_data_framec sJ | |}| j|| j|| j|dddI dH }| j||| |dI dH S )a$
Execute asynchronous Flux query and return stream of :class:`~pandas.core.frame.DataFrame` as an AsyncGenerator[:class:`~pandas.core.frame.DataFrame`].
.. note:: If the ``query`` returns tables with differing schemas than the client generates a :class:`~DataFrame` for each of them.
:param query: the Flux query
:param str, Organization org: specifies the organization for executing the query;
Take the ``ID``, ``Name`` or ``Organization``.
If not specified the default value from ``InfluxDBClientAsync.org`` is used.
:param data_frame_index: the list of columns that are used as DataFrame index
:param params: bind parameters
:param use_extension_dtypes: set to ``True`` to use panda's extension data types.
Useful for queries with ``pivot`` function.
When data has missing values, column data type may change (to ``object`` or ``float64``).
Nullable extension types (``Int64``, ``Float64``, ``boolean``) support ``panda.NA`` value.
For more info, see https://pandas.pydata.org/docs/user_guide/missing_data.html.
:return: :class:`AsyncGenerator[:class:`DataFrame`]`
.. warning:: For the optimal processing of the query results use the ``pivot() function`` which align results as a table.
.. code-block:: text
from(bucket:"my-bucket")
|> range(start: -5m, stop: now())
|> filter(fn: (r) => r._measurement == "mem")
|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
For more info see:
- https://docs.influxdata.com/resources/videos/pivots-in-flux/
- https://docs.influxdata.com/flux/latest/stdlib/universe/pivot/
- https://docs.influxdata.com/flux/latest/stdlib/influxdata/influxdb/schema/fieldsascols/
T)Zdataframe_queryr N)r$ r" r r% )r r r r Z_to_data_frame_stream_asyncr )r r r r$ r r% r" r r r r& s "
z%QueryApiAsync.query_data_frame_stream)r r c s@ | |}| j|| |||dI dH }| I dH }|tS )a
Execute asynchronous Flux query and return result as raw unprocessed result as a str.
:param query: a Flux query
:param str, Organization org: specifies the organization for executing the query;
Take the ``ID``, ``Name`` or ``Organization``.
If not specified the default value from ``InfluxDBClientAsync.org`` is used.
:param dialect: csv dialect format
:param params: bind parameters
:return: :class:`~str`
r N)r r r readdecoder )r r r dialectr resultZ raw_bytesr r r query_raw s
zQueryApiAsync.query_rawc sV | j j||ddddI d H }d|j kr2dksRn | I d H }tt||d|S )NFT)r r Z async_reqZ_preload_contentZ_return_http_data_only i+ )Z http_resp)Z
_query_apiZpost_query_asyncstatusr) r r )r r r r" datar r r r s zQueryApiAsync._post_query)NN)NN)NNNF)NNNF)__name__
__module__ __qualname____doc__r r strdictr r r r# r boolr( r& r r r- r
__classcell__r r r r r s$ P +
+r N)r4 typingr r Z influxdb_client.client._baser Z!influxdb_client.client.flux_tabler r Z influxdb_client.client.query_apir Zinfluxdb_client.restr r Z _async.restr r r r r r s