Python sqlite3 executemany 执行 SELECT 报 ProgrammingError
区分批量 DML 与查询结果收集,改用逐项 execute 或绑定参数的 IN 查询,并核对重复项、顺序和空输入。
把多个 ID 交给 sqlite3.Cursor.executemany() 执行 SELECT,会报 ProgrammingError,即使同一个游标刚刚用 executemany() 批量插入成功。错误文本可能是 You cannot execute SELECT statements in executemany(),也可能说明该接口只能执行 DML 语句。
Python 3 标准库的 sqlite3 没有把这个方法设计成“执行多个查询并合并结果”。它重复执行一条参数化的 DML 语句,不能用来取得一批 SELECT 结果。调整 fetchall() 的位置或添加 commit(),不会改变这个限制。
每个输入都要单独得到结果时,用 execute 循环
如果接口需要保留输入 ID 的顺序、重复项,以及每个 ID 是否查到记录,可以对每个参数执行查询。下面函数接收现有连接,表名与字段是固定示例:
def find_each(con, ids):
results = []
for item_id in ids:
rows = con.execute(
"SELECT id, genre FROM Genre WHERE id = ?",
(item_id,),
).fetchall()
results.append((item_id, rows))
return results
(item_id,) 是单元素参数元组;直接传一个整数不是所需的参数序列。每次 fetchall() 在执行下一次查询前完成,结果也按当前输入保存,避免最后只留下最后一次查询的结果。
这里只查询,不替调用者提交或回滚现有业务事务。多次查询是否需要看到同一份数据快照,应由应用原有的事务边界决定,不能为了修 API 调用就在函数里结束别人的事务。
只要匹配 ID 集合时,一条 IN 查询更直接
若要求只是“返回这些 ID 对应的行”,并且列表规模已确认在当前 SQLite 参数上限之内,可以生成相应数量的问号,再绑定值:
def find_set(con, ids):
ids = list(ids)
if not ids:
return []
marks = ",".join("?" for _ in ids)
sql = "SELECT id, genre FROM Genre WHERE id IN ({}) ORDER BY id".format(marks)
return con.execute(sql, ids).fetchall()
拼进 SQL 的只有程序生成的 ? 与逗号,ID 值仍通过第二个参数绑定。不要将用户输入直接拼接到 SQL,也不要指望一个 ? 自动把 Python 列表展开成多个 SQL 值。
这段和前面的逐项查询语义不同:重复的输入 ID 不会要求同一数据库行重复出现;不存在的 ID 不会自动生成空结果占位;ORDER BY id 也不是按输入列表排序。选择方案前先确定调用方究竟需要哪一种结果。
列表很大时,单条语句可能碰到变量数量限制。不能把切回 executemany() 当成大列表查询的办法;应按连接限制和结果语义设计分批或其他查询方案。这里的函数也会把输入与输出放进内存,不适合不加限制地接收无限迭代器或庞大结果。
RETURNING 也不能让 executemany 成为结果收集器
当前 sqlite3 会丢弃 executemany() 产生的结果行,包括 DML 的 RETURNING 结果。于是把 SELECT 改写成带 RETURNING 的写入,再调用 fetchall(),既改变了数据,又不能实现预期的批量结果读取。
真正需要修改并逐次读取返回行时,应按 SQLite 和 Python 的适用版本设计 execute() 与结果读取流程,并明确事务;不要为解决只读查询错误引入写入操作。executescript() 同样不是绑定参数并收集多组查询结果的替代接口。
修复后用空 ID 列表、单个已存在 ID、未命中 ID 和重复 ID 检查返回结构,确认顺序与重复规则都符合接口约定。保留原查询代码,回退时只恢复代码;若调试过程中曾额外执行写入,应按真实事务与提交记录处理,不能把一次查询异常当作所有写入都已撤销的证明。


