使用要获取的列名列表构建 psycopg2 查询

Question

一个相当简单的问题，但令人惊讶的是我们没有找到解决方案。

这是我当前的代码，用于使用psycopg2 ('2.9.1 (dt dec pq3 ext lo64)') 从 Python 3.6.9 对 PostgreSQL 数据库执行简单的 SQL 查询：

import psycopg2

myid = 100
fields = ('p.id', 'p.name', 'p.type', 'p.price', 'p.warehouse', 'p.location', )
sql_query = ("SELECT " + ', '.join(fields) + " FROM product p "
              "INNER JOIN owner o ON p.id = o.product_id "
              "WHERE p.id = {} AND (o.dateof_purchase IS NOT NULL "
              "OR o.state = 'checked_out' );"
        ).format(myid)

try:
    with psycopg2.connect(**DB_PARAMS) as conn:
        with conn.cursor(cursor_factory=DictCursor) as curs:
            curs.execute(sql_query, )
            row = curs.fetchone()

except psycopg2.Error as error:
    raise ValueError(f"ERR: something went wrong with the query :\n{sql_query}") from None

我们越来越认为这是……不好。 （说实话太糟糕了）。

因此，我们尝试使用现代 f 字符串表示法：

sql_query = (f"""SELECT {fields} FROM product p
             INNER JOIN owner o ON p.id = o.product_id
             WHERE p.id = {myid} AND (o.dateof_purchase IS NOT NULL
             OR o.state = 'checked_out' );""")

但是，查询看起来像：

SELECT  ('p.id', 'p.name', 'p.type', 'p.price', 'p.warehouse', 'p.location', ) FROM ...;

这在 PSQL 中无效，因为 1. 中的括号和 2. 中的单引号列名。

我们想想办法摆脱这些。

在此期间，我们回到文档并记住了这一点：

https://www.psycopg.org/docs/usage.html

哎呀！所以我们这样重构它：

sql_query = (f"""SELECT %s FROM product p
             INNER JOIN owner o ON p.id = o.product_id
             WHERE p.id = %s AND (o.dateof_purchase IS NOT NULL
             OR o.state = 'checked_out' );""")  

try:
    with psycopg2.connect(**DB_PARAMS) as conn:
        with conn.cursor(cursor_factory=DictCursor) as curs:
            # passing a tuple as it only accept one more argument after the query!
            curs.execute(sql_query, (fields, myid))
            row = curs.fetchone()

和mogrify() 说：

"SELECT ('p.id', 'p.name', 'p.type', 'p.price', 'p.warehouse', 'p.location', ) FROM ...;"

在这里，括号和单引号引起了麻烦，但实际上并没有引发错误。
唯一的问题是 row 计算出这个奇怪的结果：

['('p.id', 'p.name', 'p.type', 'p.price', 'p.warehouse', 'p.location', )']

那么，我们怎样才能巧妙地、动态地构建一个 psycopg2 查询，使用列名的参数列表而不忽略安全性？

（一个技巧可能是获取所有列并在之后将它们过滤掉...但是列太多了，有些包含我们不需要的大量数据，这就是为什么我们希望使用精确定义的列选择来运行查询，这些列可能会被某些函数动态扩展，否则我们当然会硬编码这些列名）。

操作系统：Ubuntu 18.04
PostgreSQL：13.3（Debian 13.3-1.pgdg100+1）

Answer 1

正如@AdamKG 所指出的，“%s”插入将尝试将每个参数转换为 SQL 字符串。相反，您可以使用 psycopg2.sql 模块将标识符插入到查询中，而不仅仅是字符串：

from psycopg2 import sql

fields = ('id', 'name', 'type', 'price', 'warehouse', 'location', )

sql_query = sql.SQL(
          """SELECT {} FROM product p
             INNER JOIN owner o ON p.id = o.product_id
             WHERE p.id = %s AND (o.dateof_purchase IS NOT NULL
             OR o.state = 'checked_out' );""")

try:
    with psycopg2.connect(**DB_PARAMS) as conn:
        with conn.cursor(cursor_factory=DictCursor) as curs:
            # passing a tuple as it only accept one more argument after the query!
            curs.execute(sql_query.format(*[sql.Identifier(field) for field in fields]), (*fields, myid))
            row = curs.fetchone()

Answer 2

我终于找到了解决办法。它利用map 来使用列名的列表或元组，并利用sql.Literal 来使用给定的id，这可能更简洁：

conn = psycopg2.connect(**DB_PARAMS)

myid = 100
# using the simple column identifiers
fields_1 = ('id', 'name', 'type', 'price', 'warehouse', 'location',)
# using the dot notation with the table alias 'p' as the prefix:
fields_2 = ('p.id', 'p.name', 'p.type', 'p.price', 'p.warehouse', 'p.location',)

sql_query_1 = sql.SQL("""
    SELECT {f} FROM product p
    INNER JOIN owner o ON p.id = o.product_id
    WHERE p.id = {j} AND (o.dateof_purchase IS NOT NULL
    OR o.state = 'checked_out' );"""
).format(
    f = sql.SQL(',').join(map(sql.Identifier, fields_1)),
    j = sql.Literal(myid)
)

sql_query_2 = sql.SQL("""
    SELECT {f} FROM product p
    INNER JOIN owner o ON p.id = o.product_id
    WHERE p.id = {j} AND (o.dateof_purchase IS NOT NULL
    OR o.state = 'checked_out' );"""
).format(
    f = sql.SQL(',').join(map(sql.SQL, fields_2)), # use sql.SQL!
    j = sql.Literal(myid)
)

sql_query_2b = sql.SQL("""
    SELECT {f} FROM product p
    INNER JOIN owner o ON p.id = o.product_id
    WHERE p.id = {j} AND (o.dateof_purchase IS NOT NULL
    OR o.state = 'checked_out' );"""
).format(
    f = sql.SQL(',').join(map(sql.Identifier, fields_2)), # DON'T use sql.Identifier!
    j = sql.Literal(myid)
)

# VALID SQL QUERY:
print(sql_query_1.as_string(conn))
# will print:
# SELECT "id","name","type","price","warehouse","location" FROM product p
#    INNER JOIN owner o ON p.id = o.product_id
#    WHERE p.id = 100 AND (o.dateof_purchase IS NOT NULL
#    OR o.state = 'checked_out' );


# VALID SQL QUERY:
print(sql_query_2.as_string(conn))
# will print:
# SELECT p.id,p.name,p.type,p.price,p.warehouse,p.location FROM product p
#    INNER JOIN owner o ON p.id = o.product_id
#    WHERE p.id = 100 AND (o.dateof_purchase IS NOT NULL
#    OR o.state = 'checked_out' );

# /!\ INVALID SQL QUERY /!\:
print(sql_query_2b.as_string(conn))
# will print:
# SELECT "p.id","p.name","p.type","p.price","p.warehouse","p.location" FROM product p
#    INNER JOIN owner o ON p.id = o.product_id
#    WHERE p.id = 100 AND (o.dateof_purchase IS NOT NULL
#    OR o.state = 'checked_out' );

但正因为如此：

• 简单的列名在双引号中被正确评估，例如。 id 相当于 "id"，name 相当于 "name" 用于 PostgreSQL，
• 列名，当使用表别名或标识符以点表示法作为前缀时，例如p.id 或 product.id 而不仅仅是 id 或 "id" 将失败并出现以下错误：

UndefinedColumn: column "p.id" does not exist
LINE 1: SELECT "p.id","p.type","p.price","p.warehouse","p.location",...
               ^
HINT:  Perhaps you meant to reference the column "p.id".