如何基于含重复ID的列表高效查询MySQL数据?
优化方案:单次查询+本地映射
你的实现每次循环都发起一次数据库请求,在ID数量多的时候会产生大量不必要的IO开销,更高效的方式是只执行一次数据库查询获取所有唯一ID的数据,再通过本地字典映射还原原列表的顺序和重复情况,代码实现如下:
from typing import List def get_sensors_by_ids(self, ids: List[int]) -> List[SensorModel]: if not ids: return [] # 提取列表中的唯一ID,减少查询范围 unique_ids = list(set(ids)) # 构造IN查询的参数占位符,避免SQL注入 placeholders = ", ".join(["%s"] * len(unique_ids)) query = f"SELECT * FROM sensor WHERE id IN ({placeholders})" # 用字典缓存查询结果,O(1)时间查找 sensor_map = {} with self.connection.cursor(dictionary=True, buffered=True) as cursor: cursor.execute(query, tuple(unique_ids)) for result in cursor.fetchall(): sensor_model = SensorDatabaseModel.parse_obj(result).to_sensor_model() sensor_map[sensor_model.id] = sensor_model # 按照原ID列表的顺序和重复次数返回对应模型 return [sensor_map.get(id) for id in ids]
核心优势
- 减少数据库交互次数:无论原ID列表有多少重复,只发起一次查询,大幅降低IO开销
- 保持原列表特性:完全保留输入列表的顺序和重复元素,和原实现的输出一致
- 查询效率更高:数据库批量查询的性能远优于多次单条查询,即使数据量不大也能体现优势
- 安全可靠:使用参数化查询避免SQL注入风险,和原实现的安全级别一致
注意事项
- 如果存在数据库中不存在的ID,
sensor_map.get(id)会返回None,和原代码中fetchone()返回None的行为保持一致;如果需要对不存在的ID抛出异常,可改为sensor_map[id] - 当输入ID列表为空时,直接返回空列表,避免无效的数据库查询
- 若
id不是数据表主键,需要确保查询结果中每个ID只返回一条数据(可在SQL中添加DISTINCT或通过业务逻辑保证)
内容的提问来源于stack exchange,提问作者Roland Deschain
相关产品推荐
相关产品推荐

