import influxdb_client

from influxdb_client import InfluxDBClient, Point from influxdb_client.client.write_api import SYNCHRONOUS bucket = "my-bucket" client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") write_api = client.write_api(write_options=SYNCHRONOUS) query_api = client.query_api() p = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3) write_api.write(bucket=bucket, record=p) ## using Table structure tables = query_api.query('from(bucket:"my-bucket") |> range(start: -10m)') for table in tables: print(table) for row in table.records: print (row.values) ## using csv library csv_result = query_api.query_csv('from(bucket:"my-bucket") |> range(start: -10m)') val_count = 0 for row in csv_result: for cell in row: val_count += 1

self.client = InfluxDBClient.from_config_file("config.ini")

self.client = InfluxDBClient.from_env_properties()

q = '''  from(bucket: stringParam)  |> range(start: -5m, stop: now())  |> filter(fn: (r) => r._measurement == "mem")  |> filter(fn: (r) => r._field == "available" or r._field == "free" or r._field == "used")  |> aggregateWindow(every: 1m, fn: mean)  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value") ''' p = { "stringParam": "my-bucket", } query_api = client.query_api(query_options=QueryOptions(profilers=["query", "operator"])) csv_result = query_api.query(query=q, params=p)

from datetime import datetime, timedelta import pandas as pd import rx from pytz import UTC from rx import operators as ops from influxdb_client import InfluxDBClient, Point, WriteOptions with InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") as _client: with _client.write_api(write_options=WriteOptions(batch_size=500, flush_interval=10_000, jitter_interval=2_000, retry_interval=5_000, max_retries=5, max_retry_delay=30_000, exponential_base=2)) as _write_client: """  Write Line Protocol formatted as string  """ _write_client.write("my-bucket", "my-org", "h2o_feet,location=coyote_creek water_level=1.0 1") _write_client.write("my-bucket", "my-org", ["h2o_feet,location=coyote_creek water_level=2.0 2", "h2o_feet,location=coyote_creek water_level=3.0 3"]) """  Write Line Protocol formatted as byte array  """ _write_client.write("my-bucket", "my-org", "h2o_feet,location=coyote_creek water_level=1.0 1".encode()) _write_client.write("my-bucket", "my-org", ["h2o_feet,location=coyote_creek water_level=2.0 2".encode(), "h2o_feet,location=coyote_creek water_level=3.0 3".encode()]) """  Write Dictionary-style object  """ _write_client.write("my-bucket", "my-org", {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"}, "fields": {"water_level": 1.0}, "time": 1}) _write_client.write("my-bucket", "my-org", [{"measurement": "h2o_feet", "tags": {"location": "coyote_creek"}, "fields": {"water_level": 2.0}, "time": 2}, {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"}, "fields": {"water_level": 3.0}, "time": 3}]) """  Write Data Point  """ _write_client.write("my-bucket", "my-org", Point("h2o_feet").tag("location", "coyote_creek").field("water_level", 4.0).time(4)) _write_client.write("my-bucket", "my-org", [Point("h2o_feet").tag("location", "coyote_creek").field("water_level", 5.0).time(5), Point("h2o_feet").tag("location", "coyote_creek").field("water_level", 6.0).time(6)]) """  Write Observable stream  """ _data = rx \ .range(7, 11) \ .pipe(ops.map(lambda i: "h2o_feet,location=coyote_creek water_level={0}.0 {0}".format(i))) _write_client.write("my-bucket", "my-org", _data) """  Write Pandas DataFrame  """ _now = datetime.now(UTC) _data_frame = pd.DataFrame(data=[["coyote_creek", 1.0], ["coyote_creek", 2.0]], index=[_now, _now + timedelta(hours=1)], columns=["location", "water_level"]) _write_client.write("my-bucket", "my-org", record=_data_frame, data_frame_measurement_name='h2o_feet', data_frame_tag_columns=['location'])

point_settings = PointSettings() point_settings.add_default_tag("id", "132-987-655") point_settings.add_default_tag("customer", "California Miner") point_settings.add_default_tag("data_center", "${env.data_center}") self.write_client = self.client.write_api(write_options=SYNCHRONOUS, point_settings=point_settings)

self.write_client = self.client.write_api(write_options=SYNCHRONOUS, point_settings=PointSettings(**{"id": "132-987-655", "customer": "California Miner"}))

self.client = InfluxDBClient.from_config_file("config.ini")

self.client = InfluxDBClient.from_env_properties()

from influxdb_client import InfluxDBClient, Point from influxdb_client.client.write_api import ASYNCHRONOUS client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") write_api = client.write_api(write_options=ASYNCHRONOUS) _point1 = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3) _point2 = Point("my_measurement").tag("location", "New York").field("temperature", 24.3) async_result = write_api.write(bucket="my-bucket", record=[_point1, _point2]) async_result.get() client.close()

from influxdb_client import InfluxDBClient, Point from influxdb_client .client.write_api import SYNCHRONOUS client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") write_api = client.write_api(write_options=SYNCHRONOUS) _point1 = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3) _point2 = Point("my_measurement").tag("location", "New York").field("temperature", 24.3) write_api.write(bucket="my-bucket", record=[_point1, _point2]) client.close()

from influxdb_client import InfluxDBClient, Point, Dialect from influxdb_client.client.write_api import SYNCHRONOUS client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") write_api = client.write_api(write_options=SYNCHRONOUS) query_api = client.query_api() """ Prepare data """ _point1 = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3) _point2 = Point("my_measurement").tag("location", "New York").field("temperature", 24.3) write_api.write(bucket="my-bucket", record=[_point1, _point2]) """ Query: using Table structure """ tables = query_api.query('from(bucket:"my-bucket") |> range(start: -10m)') for table in tables: print(table) for record in table.records: print(record.values) print() print() """ Query: using Bind parameters """ p = {"_start": datetime.timedelta(hours=-1), "_location": "Prague", "_desc": True, "_floatParam": 25.1, "_every": datetime.timedelta(minutes=5) } tables = query_api.query('''  from(bucket:"my-bucket") |> range(start: _start)  |> filter(fn: (r) => r["_measurement"] == "my_measurement")  |> filter(fn: (r) => r["_field"] == "temperature")  |> filter(fn: (r) => r["location"] == _location and r["_value"] > _floatParam)  |> aggregateWindow(every: _every, fn: mean, createEmpty: true)  |> sort(columns: ["_time"], desc: _desc) ''', params=p) for table in tables: print(table) for record in table.records: print(str(record["_time"]) + " - " + record["location"] + ": " + str(record["_value"])) print() print() """ Query: using Stream """ records = query_api.query_stream('from(bucket:"my-bucket") |> range(start: -10m)') for record in records: print(f'Temperature in {record["location"]} is {record["_value"]}') """ Interrupt a stream after retrieve a required data """ large_stream = query_api.query_stream('from(bucket:"my-bucket") |> range(start: -100d)') for record in large_stream: if record["location"] == "New York": print(f'New York temperature: {record["_value"]}') break large_stream.close() print() print() """ Query: using csv library """ csv_result = query_api.query_csv('from(bucket:"my-bucket") |> range(start: -10m)', dialect=Dialect(header=False, delimiter=",", comment_prefix="#", annotations=[], date_time_format="RFC3339")) for csv_line in csv_result: if not len(csv_line) == 0: print(f'Temperature in {csv_line[9]} is {csv_line[6]}') """ Close client """ client.close()

from influxdb_client import InfluxDBClient, Point, Dialect from influxdb_client.client.write_api import SYNCHRONOUS client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") write_api = client.write_api(write_options=SYNCHRONOUS) query_api = client.query_api() """ Prepare data """ _point1 = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3) _point2 = Point("my_measurement").tag("location", "New York").field("temperature", 24.3) write_api.write(bucket="my-bucket", record=[_point1, _point2]) """ Query: using Pandas DataFrame """ data_frame = query_api.query_data_frame('from(bucket:"my-bucket") ' '|> range(start: -10m) ' '|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value") ' '|> keep(columns: ["location", "temperature"])') print(data_frame.to_string()) """ Close client """ client.close()

""" Import VIX - CBOE Volatility Index - from "vix-daily.csv" file into InfluxDB 2.0  https://datahub.io/core/finance-vix#data """ from collections import OrderedDict from csv import DictReader import rx from rx import operators as ops from influxdb_client import InfluxDBClient, Point, WriteOptions def parse_row(row: OrderedDict): """Parse row of CSV file into Point with structure:   financial-analysis,type=ily close=18.47,high=19.82,low=18.28,open=19.82 1198195200000000000   CSV format:  Date,VIX Open,VIX High,VIX Low,VIX Close\n  2004-01-02,17.96,18.68,17.54,18.22\n  2004-01-05,18.45,18.49,17.44,17.49\n  2004-01-06,17.66,17.67,16.19,16.73\n  2004-01-07,16.72,16.75,15.5,15.5\n  2004-01-08,15.42,15.68,15.32,15.61\n  2004-01-09,16.15,16.88,15.57,16.75\n  ...   :param row: the row of CSV file  :return: Parsed csv row to [Point]  """ """  For better performance is sometimes useful directly create a LineProtocol to avoid unnecessary escaping overhead:  """ # from pytz import UTC # import ciso8601 # from influxdb_client.client.write.point import EPOCH # # time = (UTC.localize(ciso8601.parse_datetime(row["Date"])) - EPOCH).total_seconds() * 1e9 # return f"financial-analysis,type=vix-daily" \ # f" close={float(row['VIX Close'])},high={float(row['VIX High'])},low={float(row['VIX Low'])},open={float(row['VIX Open'])} " \ # f" {int(time)}" return Point("financial-analysis") \ .tag("type", "vix-daily") \ .field("open", float(row['VIX Open'])) \ .field("high", float(row['VIX High'])) \ .field("low", float(row['VIX Low'])) \ .field("close", float(row['VIX Close'])) \ .time(row['Date']) """ Converts vix-daily.csv into sequence of datad point """ data = rx \ .from_iterable(DictReader(open('vix-daily.csv', 'r'))) \ .pipe(ops.map(lambda row: parse_row(row))) client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org", debug=True) """ Create client that writes data in batches with 50_000 items. """ write_api = client.write_api(write_options=WriteOptions(batch_size=50_000, flush_interval=10_000)) """ Write data into InfluxDB """ write_api.write(bucket="my-bucket", record=data) write_api.close() """ Querying max value of CBOE Volatility Index """ query = 'from(bucket:"my-bucket")' \ ' |> range(start: 0, stop: now())' \ ' |> filter(fn: (r) => r._measurement == "financial-analysis")' \ ' |> max()' result = client.query_api().query(query=query) """ Processing results """ print() print("=== results ===") print() for table in result: for record in table.records: print('max {0:5} = {1}'.format(record.get_field(), record.get_value())) """ Close client """ client.close()

""" Efficiency write data from IOT sensor - write changed temperature every minute """ import atexit import platform from datetime import timedelta import psutil as psutil import rx from rx import operators as ops from influxdb_client import InfluxDBClient, WriteApi, WriteOptions def on_exit(db_client: InfluxDBClient, write_api: WriteApi): """Close clients after terminate a script.   :param db_client: InfluxDB client  :param write_api: WriteApi  :return: nothing  """ write_api.close() db_client.close() def sensor_temperature(): """Read a CPU temperature. The [psutil] doesn't support MacOS so we use [sysctl].   :return: actual CPU temperature  """ os_name = platform.system() if os_name == 'Darwin': from subprocess import check_output output = check_output(["sysctl", "machdep.xcpm.cpu_thermal_level"]) import re return re.findall(r'\d+', str(output))[0] else: return psutil.sensors_temperatures()["coretemp"][0] def line_protocol(temperature): """Create a InfluxDB line protocol with structure:   iot_sensor,hostname=mine_sensor_12,type=temperature value=68   :param temperature: the sensor temperature  :return: Line protocol to write into InfluxDB  """ import socket return 'iot_sensor,hostname={},type=temperature value={}'.format(socket.gethostname(), temperature) """ Read temperature every minute; distinct_until_changed - produce only if temperature change """ data = rx\ .interval(period=timedelta(seconds=60))\ .pipe(ops.map(lambda t: sensor_temperature()), ops.distinct_until_changed(), ops.map(lambda temperature: line_protocol(temperature))) _db_client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org", debug=True) """ Create client that writes data into InfluxDB """ _write_api = _db_client.write_api(write_options=WriteOptions(batch_size=1)) _write_api.write(bucket="my-bucket", record=data) """ Call after terminate a script """ atexit.register(on_exit, _db_client, _write_api) input()

""" Connect to InfluxDB 2.0 - write data and query them """ from datetime import datetime from influxdb_client import Point, InfluxDBClient from influxdb_client.client.write_api import SYNCHRONOUS """ Configure credentials """ influx_cloud_url = 'https://us-west-2-1.aws.cloud2.influxdata.com' influx_cloud_token = '...' bucket = '...' org = '...' client = InfluxDBClient(url=influx_cloud_url, token=influx_cloud_token) try: kind = 'temperature' host = 'host1' device = 'opt-123' """  Write data by Point structure  """ point = Point(kind).tag('host', host).tag('device', device).field('value', 25.3).time(time=datetime.utcnow()) print(f'Writing to InfluxDB cloud: {point.to_line_protocol()} ...') write_api = client.write_api(write_options=SYNCHRONOUS) write_api.write(bucket=bucket, org=org, record=point) print() print('success') print() print() """  Query written data  """ query = f'from(bucket: "{bucket}") |> range(start: -1d) |> filter(fn: (r) => r._measurement == "{kind}")' print(f'Querying from InfluxDB cloud: "{query}" ...') print() query_api = client.query_api() tables = query_api.query(query=query, org=org) for table in tables: for row in table.records: print(f'{row.values["_time"]}: host={row.values["host"]},device={row.values["device"]} ' f'{row.values["_value"]} °C') print() print('success') except Exception as e: print(e) finally: client.close()

from influxdb_client import InfluxDBClient _db_client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org", enable_gzip=True)

from influxdb_client import InfluxDBClient client = InfluxDBClient(url="http://localhost:8086", token="my-token") delete_api = client.delete_api() """ Delete Data """ start = "1970-01-01T00:00:00Z" stop = "2021-02-01T00:00:00Z" delete_api.delete(start, stop, '_measurement="my_measurement"', bucket='my-bucket', org='my-org') """ Close client """ client.close()

from urllib3 import Retry from influxdb_client import InfluxDBClient retries = Retry(connect=5, read=2, redirect=5) client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org", retries=retries)

from influxdb_client import Point, InfluxDBClient from influxdb_client.client.util.date_utils_pandas import PandasDateTimeHelper from influxdb_client.client.write_api import SYNCHRONOUS """ Set PandasDate helper which supports nanoseconds. """ import influxdb_client.client.util.date_utils as date_utils date_utils.date_helper = PandasDateTimeHelper() """ Prepare client. """ client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org") write_api = client.write_api(write_options=SYNCHRONOUS) query_api = client.query_api() """ Prepare data """ point = Point("h2o_feet") \ .field("water_level", 10) \ .tag("location", "pacific") \ .time('1996-02-25T21:20:00.001001231Z') print(f'Time serialized with nanosecond precision: {point.to_line_protocol()}') print() write_api.write(bucket="my-bucket", record=point) """ Query: using Stream """ query = ''' from(bucket:"my-bucket")  |> range(start: 0, stop: now())  |> filter(fn: (r) => r._measurement == "h2o_feet") ''' records = query_api.query_stream(query) for record in records: print(f'Temperature in {record["location"]} is {record["_value"]} at time: {record["_time"]}') """ Close client """ client.close()