时间戳

首页 > 解决方案 > 使用要获取的列名列表构建 psycopg2 查询

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

问题描述

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

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

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）

标签： pythonpostgresqlpsycopg2

解决方案

The '%s' insertion will try to turn every argument into an SQL string, as @AdamKG pointed out. Instead, you can use the psycopg2.sql module will allow you to insert identifiers into queries, not just strings:

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

推荐阅读