我的目标是,给定 Python 中的 id 列表,找到 id 未映射到 SQLite 表中的行。我正在尝试使用
EXCEPT
运算符来实现此目的:
-- if the table currently stores id1 and id3 would only return id2
WITH cte(id) as VALUES ('id1'), ('id2'), ('id3')
SELECT * from cte EXCEPT SELECT id FROM some_table
我想从列表中动态指定 id。我能够格式化字符串、硬编码值:
query = (
"with cte(id) as " +
f"(values {",".join(f"('{id}')" for id in ids)}) " +
"select * from cte except select id from some_table"
)
print(query)
res = cursor.execute(query)
这很容易受到 SQL 注入攻击。相反,占位符语法是首选。 Python sqlite3 文档 显示了使用
executemany
进行 INSERT
操作的示例,但是如何将其应用于 SELECT+EXCEPT 单个查询(必须使用 execute
而不是 executemany
)?或者,是否有更好的方法来通过表中不存在的输入来过滤输入列表?我的问题示例:
import sqlite3
db = sqlite3.connect(":memory:")
cursor = db.cursor()
#
# First create a table of video-id,video-title pairs
#
cursor.execute("CREATE TABLE IF NOT EXISTS videos(id TEXT PRIMARY KEY, title TEXT)")
dummy_data = [
("vid1", "Video 1"),
("vid2", "Video 2"),
("vid3", "Video 3"),
]
# use executemany to insert multiple rows via placeholder VALUES
cursor.executemany("INSERT INTO videos VALUES(?, ?)", dummy_data)
db.commit()
# sanity check that we see the expected videos
res = cursor.execute("SELECT * FROM videos")
print(f"select* result: {res.fetchall()}")
#
# Next, given a set of video ids, find all of the ids not already stored in the DB
#
new_video_ids = ["vid1", "vid2", "vid5"] # vid1 and vid2 already exist in db. only vid5 should be returned
new_video_ids_str = ",".join(f"('{id}')" for id in new_video_ids)
print(new_video_ids_str)
# The following query uses python string formatting and is therefore vulnerable to SQL injection attacks
query = (
"with cte(id) as " +
f"(values {new_video_ids_str}) " +
"select * from cte except select id from videos"
)
print(query)
res = cursor.execute(query)
print(f"filter result: {res.fetchall()}")
# I'd like to use SQLite3 placeholder values but can't figure out the syntax. The following doesn't work.
# it fails since it's trying to all of the `new_video_ids` values as a single row rather than multiple rows.
#
# query = (
# "with cte(id) as " +
# "(values (?)) " +
# "select * from cte except select id from videos"
# )
# res = cursor.execute(query, new_video_ids)
# print(f"filter result: {res.fetchall()}")
db.close()
new_video_ids = ["vid1", "vid2", "vid5"] # vid1 and vid2 already exist in db. only vid5 should be returned
new_video_ids_str = ",".join(
["(?)"] * len(new_video_ids)
)
print(new_video_ids_str)
query = (
"with cte(id) as " +
+ f"(values {new_video_ids_str}) " +
+ "select * from cte except select id from videos"
)
print(query)
res = cursor.execute(query, new_video_ids)
print(f"filter result: {res.fetchall()}")