动态 SQL WHERE 子句生成答案

【问题标题】：Dynamic SQL WHERE clause generation动态 SQL WHERE 子句生成
【发布时间】：2013-03-19 09:47:22
【问题描述】：

为了记录，我使用的是 Python 和 SQLlite。我有一个可以生成我需要的 SQL 的工作函数，但它似乎不正确。

def daily(self, host=None, day=None):
    sql = "SELECT * FROM daily WHERE 1"
    if host:
        sql += " AND host = '%s'" % (host,)
    if day:
        sql += " AND day = '%s'" % (day,)
    return sql

稍后我可能需要添加多个列和条件。

有更好的想法吗？

编辑： 看起来不正确的是我正在从字符串动态构造 SQL。这通常不是最好的方法。 SQL 注入攻击，需要正确转义字符串。我不能使用占位符，因为有些值是 None 并且不需要在 WHERE 子句条件中。

【问题讨论】：

标签： python sql sqlite language-agnostic sql-injection

【解决方案1】：

您真的不想使用字符串格式来包含值。通过 SQL 参数将其留给数据库 API。

使用你的参数：

让数据库有机会准备语句并重用查询计划以获得更好的性能。
免去正确转义值的麻烦（包括避免允许 SQL 转义和那些 SQL 注入攻击）。

由于 SQLLite supports named SQL parameters，我将返回一个语句和一个带参数的字典：

def daily(self, host=None, day=None):
    sql = "SELECT * FROM daily"
    where = []
    params = {}
    if host is not None:
        where.append("host = :host")
        params['host'] = host
    if day is not None:
        where.append("day = :day")
        params['day'] = day
    if where:
        sql = '{} WHERE {}'.format(sql, ' AND '.join(where))
    return sql, params

然后将 both 传递给cursor.execute():

cursor.execute(*daily(host, day))

SQL 生成变得复杂快，你可能想看看SQLAlchemy core 来代替。

对于您的示例，您可以生成：

from sqlalchemy import Table, Column, Integer, String, Date, MetaData

metadata = MetaData()
daily = Table('daily', metadata, 
    Column('id', Integer, primary_key=True),
    Column('host', String),
    Column('day', Date),
)
from sqlalchemy.sql import select

def daily(self, host=None, day=None):
    query = select([daily])
    if host is not None:
        query = query.where(daily.c.host == host)
    if day is not None:
        query = query.where(daily.c.day == day)
    return query

query 对象可以应用其他过滤器、排序、分组、用作其他查询的子选择、连接并最终发送以执行，此时 SQLAlchemy 会将其转换为适合您的特定数据库的 SQL正在连接。

【讨论】：

感谢 SQLAlchemy 链接。但它超出了范围。仅限标准/内置模块限制。
@Ayman：那么至少使用 SQL 参数。为什么会有这么随意的限制？
与第三方图书馆合作的企业政策。我也讨厌这样，但对此无能为力。
它受到 MIT 许可证的保护，几乎不会成为病毒式的责任。我很同情不得不在这样一个限制性的环境中工作！

【解决方案2】：

只是为了完整性。我发现 pypika 库非常方便（如果允许使用库）：

https://pypika.readthedocs.io/en/latest/index.html

它允许像这样构造 sql 查询：

from pypika import Query

q = Query._from('daily').select('*')
if host:
     q = q.where('host' == host)
if day:
     q = q.where('day' == day)
sql = str(q)

【讨论】：