如何用Python将API响应数据插入QuestDB并解决解析报错

需要编写Python脚本，将API返回的包含timeStamp和value字段的时序数据插入QuestDB。通过socket连接发送拼接字符串时，出现could not parse measurement, INCOMPLETE_TAG解析错误，需提供正确实现并解决该报错。

测试数据示例：

b"[{'timeStamp': '2019-02-01T01:00:00', 'value': 1794.0}, {'timeStamp': '2019-03-01T01:00:00', 'value': 1486.0}, ...]"

现有错误代码：

def main_request(extractUrl,headers): 
        #using requests to call API data 
        r = requests.get(extractUrl,headers=headers) 
        #returning data in json format 
        return r.json() 

if __name__ == '__main__':
    try:
        while 1==1:
            # Get data from the API variable calling main function 
            data_month = main_request(monthUrl, headers=headersAPI)
            
            # creating a dataframe using pandas 
            data = str(data_month['measurements']) 
            
            # Send data to QuestDB
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((HOST, PORT))
            #sock.sendall(data)
            sock.sendall(('measurements,'+data).encode())

            sleep(2)

    except Exception as e:
        print(e)

现有代码直接将JSON数组转为字符串拼接，完全不符合QuestDB兼容的InfluxDB Line Protocol格式。该错误提示说明QuestDB无法解析你发送的字符串，因为格式完全不匹配。

Line Protocol的基本格式为：

measurement_name[,tag_key=tag_value...] field_key=field_value [timestamp]

• measurement_name：表名（对应你用的measurements）
• 标签（可选）：键值对，用于维度过滤，格式key=value，多个用逗号分隔
• 字段：必填，数值型字段直接写key=数值，字符串要加引号
• 时间戳：可选，需为纳秒级整数，如果不填QuestDB会自动用当前时间

import requests
import socket
import time
from datetime import datetime

def main_request(extractUrl, headers): 
    r = requests.get(extractUrl, headers=headers) 
    return r.json() 

def convert_to_line_protocol(measurements):
    lines = []
    for item in measurements:
        # 将ISO时间字符串转为纳秒级时间戳
        dt = datetime.fromisoformat(item['timeStamp'])
        timestamp = int(dt.timestamp() * 1e9)
        # 构造Line Protocol字符串：表名 字段=值 时间戳
        line = f"measurements value={item['value']} {timestamp}"
        lines.append(line)
    # 每个数据点用换行分隔，QuestDB支持批量提交
    return '\n'.join(lines) + '\n'

if __name__ == '__main__':
    # 配置QuestDB的socket地址（默认Line Protocol端口是9009）
    HOST = 'localhost'
    PORT = 9009
    monthUrl = "你的API地址"
    headersAPI = {"你的请求头"}

    try:
        while True:
            data_month = main_request(monthUrl, headers=headersAPI)
            measurements = data_month['measurements']
            # 转换为符合要求的Line Protocol格式
            line_data = convert_to_line_protocol(measurements)
            
            # 发送数据到QuestDB
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((HOST, PORT))
            sock.sendall(line_data.encode('utf-8'))
            sock.close()  # 关闭socket，避免资源泄漏
            
            print(f"已成功插入{len(measurements)}条数据")
            time.sleep(2)

    except Exception as e:
        print(f"错误：{e}")

• 实现了convert_to_line_protocol函数，将每个数据点转为标准Line Protocol格式
• 将timeStamp转换为纳秒级时间戳，符合QuestDB的时间格式要求
• 每个数据点单独构造一行，批量提交时用换行分隔
• 发送后关闭socket，避免长时间占用连接资源
• 移除了无意义的str(data_month['measurements'])操作，直接处理原始JSON数组

内容的提问来源于stack exchange，提问作者Mikael Rönn