动态 SQL WHERE 子句生成

Posted 2023-03-24

【中文标题】动态 SQL WHERE 子句生成

【英文标题】：Dynamic SQL WHERE clause generation

【发布时间】：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 子句条件中。

【参考方案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)