SQLite“IN”子句的参数替换答案

【问题标题】：Parameter substitution for a SQLite "IN" clauseSQLite“IN”子句的参数替换
【发布时间】：2010-11-21 13:15:52
【问题描述】：

我正在尝试将参数替换与SQLite within Python 一起用于 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]你不应该使用 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
        ) 
    )

【问题讨论】：

感谢您的所有回答。当我终于看到我只需要为我要替换的每个参数添加一个问号时，这很有意义。

标签： python sqlite

【解决方案1】：

您确实需要正确数量的?s，但这不会造成 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')]

【讨论】：

+1 用于生成占位符列表字符串的“最佳”解决方案 :-)
有没有一种简单的方法可以改为使用命名参数？像:id1 :id2 :id3 之类的东西。我在带有一些其他命名参数的更大查询的上下文中使用它。
几年后我会谈到这个，但我也需要命名参数。我刚刚这样做了：query = "SELECT * FROM my_table WHERE my_param = :my_param AND id IN ({})".format(', '.join(':{}'.format(i) for i in range(len(desired_ids)))) ; params = {'my_param': 'foo'} ; params.update({str(i): id for i, id in enumerate(desired_ids)}) ; result = cursor.execute(query, params) sqlite3 模块非常满意 :0、:1、:2 之类的字符串替换参数。（堆栈溢出确实破坏了 cmets 中的代码格式；抱歉，这很难阅读。）
感谢您的回答，geekofalltrades。我真的希望这可能是一个清晰的顶级答案，与其他无法处理命名参数的答案一起使用。

【解决方案2】：

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

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

【讨论】：

现在根据同一个链接，显然这个限制已增加到“3.32.0 之后的 SQLite 版本为 32766”。

【解决方案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中的每个元素都有一个问号。

【讨论】：

这是一个很好的解释。谢谢。

【解决方案4】：

我总是做这样的事情：

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

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

【讨论】：

我将在 IN 子句中使用的值实际上来自从另一个系统导出的文件。我预计注射的风险很小，但你永远不知道 Bobby Tables 什么时候会出现。
injecton 的风险为 0，因为您以编程方式输入查询的唯一内容是一堆问号。假设的攻击者所能做的就是控制问号的数量——这不是攻击向量。实际的外部提供的数据是通过 ?像往常一样的参数传递机制。

【解决方案5】：

您可以使用非常薄的层，例如 notanorm

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

...然后您的代码如下所示：

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'}

HN：https://news.ycombinator.com/item?id=26733028

【讨论】：

【解决方案6】：

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

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_id0, :desired_id1, :desired_id2, :desired_id3) AND type IN (:desired_type0, :desired_type1)

# # and
# sql_dict = {
#     "some": "other parameters you might need",
#     "desired_id0": "1",
#     "desired_id1": "2",
#     "desired_id2": "5",
#     "desired_id3": "47",
#     "desired_type0": "active",
#     "desired_type1": "inactive",
# }

【讨论】：

【解决方案7】：

如果 sqlite 对 sql 请求的长度有问题，则不定数量的问号可能是某种方式。

【讨论】：