Python使用requests调用InfluxDB接口返回CSV如何按时间戳解析排序

实现方案 #

你可以直接用Python内置的csv模块解析返回的CSV格式数据，配合datetime转换时间戳后完成排序，完整可运行的代码如下：

import requests
import csv
from io import StringIO
from datetime import datetime
import json

def get_Oversightresponse():
    url = "http://172.16.26.252:8090/api/v2/query?org=mega&bucket=monitoring"
    payload="""from(bucket: "monitoring")
|> range(start: -60m, stop: now())
|> filter(fn: (r) => r["_measurement"] == "snmp_trap")
|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn:"_value")
|> map(fn: (r) => ({
time: r._time,
level: r["engine.35.1.14.1.1"],
name: r["engine.35.1.14.1.2"],
signaler: r["engine.35.1.14.1.3"],
payload: r["engine.35.1.14.1.6"],
}))
|> filter(fn: (r) => r.level > 30 and r.name != "signal-fetch-failed")"""

    headers = {'Authorization': 'Token secrettoken','Accept': 'application/csv','Content-Type': 'application/vnd.flux'}
    response = requests.request("POST", url, headers=headers, data=payload)

    if response.status_code // 100 == 2:
        # 包装返回文本为文件对象供csv模块读取
        csv_data = StringIO(response.text)
        reader = csv.DictReader(csv_data)
        records = []
        for row in reader:
            # 转换各字段类型方便后续使用
            processed_row = {
                "level": int(row["level"]),
                "name": row["name"],
                "signaler": row["signaler"],
                # 转换时间为datetime对象用于排序，兼容低版本Python
                "time": datetime.fromisoformat(row["time"].replace("Z", "+00:00")),
                # 自动解析payload的JSON字符串为字典
                "payload": json.loads(row["payload"])
            }
            records.append(processed_row)
        
        # 按时间升序排序，需要倒序的话添加reverse=True参数即可
        sorted_records = sorted(records, key=lambda x: x["time"])
        
        # 测试输出排序结果
        for item in sorted_records:
            print(f"时间:{item['time']} | 告警名称:{item['name']} | 级别:{item['level']}")
        return sorted_records
    else:
        print(f"请求错误，状态码:{response.status_code}")
        return None

补充说明 #

• 代码兼容Python 3.7及以上所有版本
• 解析后的数据已经全部转换为对应Python原生类型，可直接读取使用
• 如需调整排序规则，只需修改sorted函数的key参数即可

内容的提问来源于stack exchange，提问作者Asad Javed