SQLite“IN”的参数替换条款

Question

我正在尝试使用 Python 中的 SQLite 作为 IN 子句的参数替换。这是一个完整的运行示例，演示了：

import sqlite3

c = sqlite3.connect(":memory:")
c.execute('CREATE TABLE distro (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)')

for name in 'Ubuntu Fedora Puppy DSL SuSE'.split():
  c.execute('INSERT INTO distro (name) VALUES (?)', [ name ] )

desired_ids = ["1", "2", "5", "47"]
result_set = c.execute('SELECT * FROM distro WHERE id IN (%s)' % (", ".join(desired_ids)), ())
for result in result_set:
  print result

它打印出：

(1, u'Ubuntu') （2，u'Fedora'） （5，u'SuSE'）

由于文档指出“[y]ou 不应该使用 Python 的字符串操作来组装查询，因为这样做是不安全的；它使您的程序容易受到 SQL 注入攻击”，我希望使用参数替换。

当我尝试：

result_set = c.execute('SELECT * FROM distro WHERE id IN (?)', [ (", ".join(desired_ids)) ])

我得到一个空结果集，当我尝试：

result_set = c.execute('SELECT * FROM distro WHERE id IN (?)', [ desired_ids ] )

我得到：

InterfaceError：绑定参数 0 时出错 - 可能不受支持的类型。

虽然我希望这个简化问题的任何答案都能起作用，但我想指出我想要执行的实际查询是在双重嵌套子查询中。也就是说：

UPDATE dir_x_user SET user_revision = user_attempted_revision 
WHERE user_id IN 
    (SELECT user_id FROM 
        (SELECT user_id, MAX(revision) FROM users WHERE obfuscated_name IN 
            ("Argl883", "Manf496", "Mook657") GROUP BY user_id
        ) 
    )

Answer 1

您确实需要正确数量的 ?，但这不会造成 sql 注入风险：

>>> result_set = c.execute('SELECT * FROM distro WHERE id IN (%s)' %
                           ','.join('?'*len(desired_ids)), desired_ids)
>>> print result_set.fetchall()
[(1, u'Ubuntu'), (2, u'Fedora'), (5, u'SuSE')]

Answer 2

根据http://www.sqlite.org/limits.html（第9项），SQLite无法（默认情况下）处理超过 999 个查询参数，因此如果您要IN查找数千个项目，此处的解决方案（生成所需的占位符列表）将会失败。如果是这种情况，您将需要分解列表，然后循环遍历其中的各个部分并自行连接结果。

如果您的 IN 子句中不需要数千个项目，那么 Alex 的解决方案就是实现这一目标的方法（并且似乎是 Django 的实现方式）。

Answer 3

更新：这有效：

import sqlite3

c = sqlite3.connect(":memory:")
c.execute('CREATE TABLE distro (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)')

for name in 'Ubuntu Fedora Puppy DSL SuSE'.split():
  c.execute('INSERT INTO distro (name) VALUES (?)', ( name,) )

desired_ids = ["1", "2", "5", "47"]
result_set = c.execute('SELECT * FROM distro WHERE id IN (%s)' % ("?," * len(desired_ids))[:-1], desired_ids)
for result in result_set:
  print result

问题是你需要一个？对于输入列表中的每个元素。

语句 ("?," * len(desired_ids))[:-1] 生成一个重复的“?,”字符串，然后截掉最后一个逗号。这样，desired_ids 中的每个元素都有一个问号。

Answer 4

我总是最终做这样的事情：

query = 'SELECT * FROM distro WHERE id IN (%s)' % ','.join('?' for i in desired_ids)
c.execute(query, desired_ids)

不存在注入风险，因为您没有将desired_ids中的字符串直接放入查询中。

Answer 5

我需要使用一些其他命名参数，因此我开发了两个可能值得分享的辅助函数。

def prepare_list_query(name, values):
    """Prepare SQLite query with named parameters."""
    list_query = ", ".join(":{}_{}".format(name, i) for i in range(len(values)))
    return list_query


def prepare_list_dict(name, values):
    """Prepare SQLite dict with named parameters."""
    list_dict = {"{}_{}".format(name, i): value for i, value in enumerate(values)}
    return list_dict

# Usage:

desired_ids = ["1", "2", "5", "47"]
desired_types = ["active", "inactive"]

sql = "SELECT * FROM distro WHERE id IN ({}) AND type IN ({})".format(
    prepare_list_query("desired_id", desired_ids),
    prepare_list_query("desired_type", desired_types),
)
sql_dict = {"some": "other parameters you might need"}
sql_dict.update(prepare_list_dict("desired_id", desired_ids))
sql_dict.update(prepare_list_dict("desired_type", desired_types))

# # This results in:
# sql = "SELECT * FROM distro WHERE id IN (:desired_id_0, :desired_id_1, :desired_id_2, :desired_id_3) AND type IN (:desired_type_0, :desired_type_1)

# # and
# sql_dict = {
#     "some": "other parameters you might need",
#     "desired_id_0": "1",
#     "desired_id_1": "2",
#     "desired_id_2": "5",
#     "desired_id_3": "47",
#     "desired_type_0": "active",
#     "desired_type_1": "inactive",
# }

# And now execute the query:
result = c.execute(sql, sql_dict)

Answer 6

我编写了一个名为 notanorm 的非常薄的层来解决这个问题。

https://pypi.org/project/notanorm/

...然后你的代码看起来像this：

import notanorm

c = notanorm.SqliteDb(":memory:")
c.query('CREATE TABLE distro (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)')

for name in 'Ubuntu Fedora Puppy DSL SuSE'.split():
  c.insert('distro', name=name)

desired_ids = ["1", "2", "5", "47"]
result_set = c.select('distro', id=desired_ids)

for result in result_set:
  print(result)

这既易于阅读，又允许您有一天切换数据库。

{'id': 1, 'name': 'Ubuntu'}
{'id': 2, 'name': 'Fedora'}
{'id': 5, 'name': 'SuSE'}

还可以考虑查看较重的框架： https://www.sqlalchemy.org/