types-influxdb-client

import os from influxdb_client import InfluxDBClient, Point from influxdb_client.client.write_api import SYNCHRONOUS # Configuration (use environment variables or replace with actual values) INFLUXDB_URL = os.environ.get('INFLUXDB_URL', 'http://localhost:8086') INFLUXDB_TOKEN = os.environ.get('INFLUXDB_TOKEN', 'my-super-secret-token') INFLUXDB_ORG = os.environ.get('INFLUXDB_ORG', 'my-org') INFLUXDB_BUCKET = os.environ.get('INFLUXDB_BUCKET', 'my-bucket') def write_and_query_data() -> None: try: with InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG) as client: write_api = client.write_api(write_options=SYNCHRONOUS) p = Point("my_measurement").tag("location", "Prague").field("temperature", 25.3) write_api.write(bucket=INFLUXDB_BUCKET, record=p) print(f"Successfully wrote point: {p.to_line_protocol()}") query_api = client.query_api() tables = query_api.query(f'from(bucket:"{INFLUXDB_BUCKET}") |> range(start: -1h)') print("\nQuery Results:") for table in tables: for record in table.records: print(f" {record.get_measurement()}: {record.get_field()}={record.get_value()} @ {record.get_time()}") except Exception as e: print(f"An error occurred: {e}") if __name__ == "__main__": write_and_query_data()