问题描述

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

以下是我当前的代码，用于使用psycopg2('2.9.1 (dt dec pq3 ext lo64)')在来自Python3.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 :
{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', )']

那么，我们如何才能在不忽略安全性的情况下，使用列名的参数列表巧妙而动态地构建一个eeCopg2查询呢？

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

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()