在 psycopg2 中构建动态更新查询答案

【问题标题】：Build a dynamic update query in psycopg2在 psycopg2 中构建动态更新查询
【发布时间】：2016-01-13 18:33:30
【问题描述】：

我必须为 postgresql 构造一个动态更新查询。它是动态的，因为我必须事先确定要更新哪些列。

给定一个示例表：

create table foo (id int, a int, b int, c int)

然后我将以编程方式构造“set”子句

_set = {}
_set['a'] = 10
_set['c'] = NULL

之后我必须构建更新查询。在这里我被困住了。我必须构造这个 sql 更新命令：

update foo set a = 10, b = NULL where id = 1

如何使用 psycopg2 参数化命令执行此操作？（即，如果字典不为空，则循环遍历并构建 set 子句）？

更新

在我睡觉的时候，我自己找到了解决方案。它是动态的，正是我想要的样子:-)

create table foo (id integer, a integer, b integer, c varchar)

updates = {}
updates['a'] = 10
updates['b'] = None
updates['c'] = 'blah blah blah'
sql = "upgrade foo set %s where id = %s" % (', '.join("%s = %%s" % u for u in updates.keys()), 10)
params = updates.values()
print cur.mogrify(sql, params)
cur.execute(sql, params)

结果就是我需要什么以及如何需要（尤其是可空和可引用的列）：

"upgrade foo set a = 10, c = 'blah blah blah', b = NULL where id = 10"

【问题讨论】：

睡觉时？ :)

标签： python postgresql psycopg2

【解决方案1】：

实际上有一种更清洁的方法，使用the alternative column-list syntax：

sql_template = "UPDATE foo SET ({}) = %s WHERE id = {}"
sql = sql_template.format(', '.join(updates.keys()), 10)
params = (tuple(addr_dict.values()),)
print cur.mogrify(sql, params)
cur.execute(sql, params)

【讨论】：

这很好，但我也建议使用 %s 作为 id 值，这样您就可以防止使用 psycopg2 进行 sql 注入 - 以防万一该值来自某个未知客户端
真实故事 phouse！
谢谢，这正是我所需要的。奇怪的是，记录如此之少。在深入研究之后，它似乎是主要 SQL 规范的一部分，应该可以在任何符合规范的 SQL 语言中使用。

【解决方案2】：

使用psycopg2.sql——SQL字符串组合模块

该模块包含用于以方便和安全的方式动态生成 SQL 的对象和函数。

from psycopg2 import connect, sql

conn = connect("dbname=test user=postgres")

upd = {'name': 'Peter', 'age': 35, 'city': 'London'}
ref_id = 12

sql_query = sql.SQL("UPDATE people SET {data} WHERE id = {id}").format(
    data=sql.SQL(', ').join(
        sql.Composed([sql.Identifier(k), sql.SQL(" = "), sql.Placeholder(k)]) for k in upd.keys()
    ),
    id=sql.Placeholder('id')
)
upd.update(id=ref_id)
with conn:
    with conn.cursor() as cur:
        cur.execute(sql_query, upd)
conn.close()

在关闭连接之前运行print(sql_query.as_string(conn)) 将显示此输出：

UPDATE people SET "name" = %(name)s, "age" = %(age)s, "city" = %(city)s WHERE id = %(id)s

【讨论】：

【解决方案3】：

不需要动态 SQL。假设a 不可为空，而b 可空。

如果您想同时更新a 和b：

_set = dict(
    id = 1,
    a = 10,
    b = 20, b_update = 1
)
update = """
    update foo
    set
        a = coalesce(%(a)s, a), -- a is not nullable
        b = (array[b, %(b)s])[%(b_update)s + 1] -- b is nullable
    where id = %(id)s
"""
print cur.mogrify(update, _set)
cur.execute(update, _set)

输出：

update foo
set
    a = coalesce(10, a), -- a is not nullable
    b = (array[b, 20])[1 + 1] -- b is nullable
where id = 1

如果你不想更新：

_set = dict(
    id = 1,
    a = None,
    b = 20, b_update = 0
)

输出：

update foo
set
    a = coalesce(NULL, a), -- a is not nullable
    b = (array[b, 20])[0 + 1] -- b is nullable
where id = 1

【讨论】：

不幸的是，我必须使用动态查询，因为我的真实表有 30 多列，我只想更新那些已更新的列（如果有的话）。当然，我必须处理应该引用的 varchar 列。但我不能自动引用每个值，因为例如可空列的值将变为“无”或“空”
@Gabor：我想你不明白它是如何工作的。我稍后或明天会回复。
好的。我正在寻找解决方案。但我不想在 _set 字典或更新字符串中获得完整的列数。当然，我必须注意引用的列，就像它是一个可为空的 varchar 列一样。

【解决方案4】：

一个不使用 python 格式的选项，使用 psycopg2 的 AsIs 函数作为列名（尽管这不会阻止您对列名进行 SQL 注入）。字典命名为data:

update_statement = f'UPDATE foo SET (%s) = %s WHERE id_column=%s'
columns = data.keys()
values = [data[column] for column in columns]
query = cur.mogrify(update_statement, (AsIs(','.join(columns)), tuple(values), id_value))

【讨论】：