Python使用influxdb_client异步写入InfluxDB的错误检测及write_api初始化问题

你使用的异步写入模式下，write_api.write() 方法会返回一个asyncio.Future对象，你可以通过两种方式捕获错误：

• 方式1：await调用+异常捕获
  直接在调用write时加await，用try-except块捕获写入过程抛出的异常，示例代码：

from influxdb_client import InfluxDBClient, WritePrecision
from influxdb_client.client.write_api import ASYNCHRONOUS
from influxdb_client.rest import ApiException

client = InfluxDBClient(url="你的influx地址", token="你的token", org="你的org")
write_api = client.write_api(write_options=ASYNCHRONOUS)

try:
    # 等待写入操作完成
    await write_api.write(bucket="你的bucket", org="你的org", data=data, write_precision=WritePrecision.US)
except ApiException as e:
    # 处理InfluxDB返回的接口错误，比如鉴权失败、bucket不存在、数据格式错误等
    print(f"写入接口错误: {e.status} {e.response}")
except Exception as e:
    # 处理其他异常，比如网络超时、本地数据序列化错误等
    print(f"写入未知错误: {str(e)}")

• 方式2：配置全局错误回调
  初始化write_api的时候可以在write_options里传入自定义错误回调函数，所有异步写入产生的错误都会自动触发该回调，不需要每次write都单独捕获：

def write_error_callback(exception, data):
    print(f"写入失败，失败数据: {data}, 错误信息: {str(exception)}")

write_api = client.write_api(write_options=ASYNCHRONOUS, error_callback=write_error_callback)

不需要每次写入都重新初始化，同一个write_api实例可以全局复用。

• 官方设计上write_api会内置维护连接池、异步写入队列、批处理配置等资源，重复初始化会造成不必要的资源开销，还可能导致队列中的未写入数据丢失。
• 仅需要注意在程序退出前，主动调用await write_api.close()，强制flush所有队列中还未发送的写入数据，避免数据丢失。

内容的提问来源于stack exchange，提问作者Rony Tesler