Flask接收JSON遥测数据应先检查request.is_json再调用get_json(),未匹配时用request.values.to_dict()兜底;嵌套字段需手动提取,避免依赖自动绑定。

Flask 路由怎么接收 JSON 遥测数据才不丢字段
直接用 request.get_json() 有时会返回 None,尤其当前端没设 Content-Type: application/json 或用了 FormData 发送。必须显式检查请求头和解析结果。
- 加
force=True强制解析(仅限可信来源):request.get_json(force=True) - 更稳妥的做法是先判断:
if request.is_json:再调用get_json(),否则用request.values.to_dict()兜底 - 遥测数据常含嵌套结构(如
{"sensor_id": "A01", "metrics": {"temp": 23.4, "hum": 65}}),Flask 默认不递归展开,需手动提取字段,别依赖自动绑定
InfluxDB 2.x 写入时 timestamp 怎么对齐才不乱序
InfluxDB 对时间戳精度和顺序敏感,写入时若时间戳重复或倒退,可能导致数据被丢弃或排序异常——尤其在高频采集(>10Hz)或设备本地时钟未同步场景下。
- 服务端统一用
time.time_ns()生成纳秒级时间戳,避免依赖客户端时间 - 写入前用
Point().time()显式设置时间,不要省略;若用 batch 写入,确保每条 Point 的时间戳严格递增 - InfluxDB 2.x 默认保留策略(retention policy)可能自动删老数据,检查
influx bucket list确认 retention duration 是否足够长(如72h不够存一周数据)
Flask + InfluxDB 查询时如何避免 timeout 和内存爆掉
时序查询常因时间范围过大、字段过多或未加 limit 导致 HTTP 超时或 Python 进程 OOM。InfluxDB 的 Flux 查询默认不限制结果集大小,Flask 默认也没有流式响应机制。
- Flux 查询里必须加
|> limit(n: 1000),且用|> range(start: -1h)替代宽泛的start: 2020-01-01 - 用
query_api.query_stream()替代query_api.query(),逐行 yield 结果,避免一次性加载全部 record 到内存 - Flask 返回前做类型转换:Flux 返回的
Record对象不能直接 jsonify,需转成 dict 并把record.get_time()转为 ISO 格式字符串
为什么 influxdb-client-python 的 write_api() 默认不是线程安全的
Flask 多 worker(如 gunicorn 启动 4 个进程)下,若在模块顶层创建单例 write_api,多个请求会并发调用同一实例,引发连接复用冲突或缓冲区竞争,表现是部分数据写入失败但无报错。
立即学习“Python免费学习笔记(深入)”;
- 不要全局初始化
write_api,改在请求内按需获取:client.write_api(write_options=WriteOptions(batch_size=1000)) - 若追求性能,可用 Flask 应用上下文绑定连接池,但必须确保每个请求获得独立
write_api实例,而非共享 - 写入失败时,
write_api不抛异常而是静默重试(默认重试 3 次),需主动检查write_api._retry_count或捕获InfluxDBError日志


















