Python怎样操作InfluxDB？influxdb-client

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

# 替换为你的 InfluxDB 配置
token = "YOUR_INFLUXDB_TOKEN"
org = "YOUR_INFLUXDB_ORG"
bucket = "YOUR_INFLUXDB_BUCKET"
url = "YOUR_INFLUXDB_URL" # 例如：http://localhost:8086

client = InfluxDBClient(url=url, token=token, org=org)

# 写入数据
write_api = client.write_api(write_options=SYNCHRONOUS)

# 创建一个数据点
point = Point("measurement_name").tag("tag_key", "tag_value").field("field_key", 123.45)

# 写入数据点
write_api.write(bucket=bucket, org=org, record=point)

# 查询数据
query_api = client.query_api()

query = f'''
from(bucket:"{bucket}")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "measurement_name")
'''

result = query_api.query(org=org, query=query)

# 处理查询结果
for table in result:
    for record in table.records:
        print(record)

# 关闭客户端
client.close()

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import time
import logging

# 配置日志
logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

# 替换为你的 InfluxDB 配置
token = "YOUR_INFLUXDB_TOKEN"
org = "YOUR_INFLUXDB_ORG"
bucket = "YOUR_INFLUXDB_BUCKET"
url = "YOUR_INFLUXDB_URL" # 例如：http://localhost:8086

max_retries = 3
retry_delay = 5  # seconds

for attempt in range(max_retries):
    try:
        client = InfluxDBClient(url=url, token=token, org=org)
        # 检查连接是否成功
        if client.ping():
            print("Connected to InfluxDB successfully!")
            break  # 连接成功，跳出循环
        else:
            raise Exception("InfluxDB ping failed.")
    except Exception as e:
        logging.error(f"Attempt {attempt + 1} failed: {e}")
        if attempt < max_retries - 1:
            print(f"Retrying in {retry_delay} seconds...")
            time.sleep(retry_delay)
    else:
        # 连接成功后，执行后续操作
        write_api = client.write_api(write_options=SYNCHRONOUS)
        point = Point("measurement_name").tag("tag_key", "tag_value").field("field_key", 123.45)
        try:
            write_api.write(bucket=bucket, org=org, record=point)
            print("Data written successfully!")
        except Exception as e:
            logging.error(f"Write operation failed: {e}")
        finally:
            client.close() # 确保关闭连接
        break # 写入成功，跳出循环
else:
    print("Failed to connect to InfluxDB after multiple retries.")

client.ping()

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import time

# 替换为你的 InfluxDB 配置
token = "YOUR_INFLUXDB_TOKEN"
org = "YOUR_INFLUXDB_ORG"
bucket = "YOUR_INFLUXDB_BUCKET"
url = "YOUR_INFLUXDB_URL"

client = InfluxDBClient(url=url, token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

# 准备一批数据点
points = []
for i in range(1000):
    point = Point("measurement_name").tag("batch", "true").field("value", i)
    points.append(point)

# 批量写入
start_time = time.time()
write_api.write(bucket=bucket, org=org, record=points)
end_time = time.time()

print(f"写入1000个数据点耗时: {end_time - start_time:.4f} 秒")

# 关闭客户端
client.close()

from influxdb_client import InfluxDBClient

# 替换为你的 InfluxDB 配置
token = "YOUR_INFLUXDB_TOKEN"
org = "YOUR_INFLUXDB_ORG"
bucket = "YOUR_INFLUXDB_BUCKET"
url = "YOUR_INFLUXDB_URL"

client = InfluxDBClient(url=url, token=token, org=org)
query_api = client.query_api()

# 使用Flux查询数据
flux_query = f'''
from(bucket:"{bucket}")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "measurement_name" and r.batch == "true")
  |> mean()
'''

result = query_api.query(org=org, query=flux_query)

# 处理查询结果
for table in result:
    for record in table.records:
        print(record)

# 关闭客户端
client.close()