在Python中使用字典将参数传递给postgresql语句[重复]

Question

我定义了一个字典，其中包含几个参数及其值，最终将用于构建 SQL 查询

query_params = collections.OrderedDict(
        {'table_name':'publilc.churn_data',
         'date_from':'201712',
         'date_to':'201805',
         'class_target':'NPA'
      })

参数将在以下查询中使用：

sql_data_sample = str("""select * from %s # get value of table_name
                                    where dt = %s    #get value of date_from
                                    and target in ('ACTIVE')

                        ----------------------------------------------------
                        union all
                        ----------------------------------------------------
                        (select * from %s #get value of table_name
                                 where dt = %s #get value of date_to
                                 and target in (%s));""") #get value of class_target
                                    %("'"+.join(str(list(query_params.values())[0])) + "'" + 
                                    "'"+.join(list(query_params.values())[1]) + "'" + 
                                    "'"+.join(list(query_params.values())[2]) + "'" +
                                    "'"+.join(list(query_params.values())[3]) + "'" )

但是这给了我一个缩进错误，如下所示：

get_ipython().run_line_magic('("\'"+.join(list(query_params.values())[0])', '+ "\'"')
    ^
IndentationError: unexpected indent

查询最终应如下所示：

select *from public.churn_data
        where dt = '201712'
        and target in ('ACTIVE')

----------------------------------------------------
union all
----------------------------------------------------
 (select * from public.churn_data 
            where dt = '201805'
            and target in ('NPA'));

我无法弄清楚错误的根源在哪里。是否是因为公众。在表名中？ 有人可以帮我解决这个问题吗？？

Answer 1

请使用参数化查询，如文档

中所述

既然你已经有一个字典，你可以这样做：

sql_data_sample = """select * from %(table_name)s
           where dt = %(date_from)s
           and target in ('ACTIVE')
           ----------------------------------------------------
           union all
           ----------------------------------------------------
           (select * from %(table_name)s
           where dt = %(date_to)s
           and target in (%(class_target)s));"""

cur.execute(sql_data_sample, query_params)

我还没有测试 if if 是否适用于有序字典，但我认为应该。如果没有，您可以将有序字典设置为常规字典，然后再将其作为参数映射传递。

编辑 除非您稍后需要参数成为 OrderedDict，否则请使用常规字典。据我所知，您只选择了 OrderedDict 来保留

list(query_params.values())[0]

的值顺序。

EDIT2 表名称和字段名称无法使用绑定传递。 Antoine Dusséaux 在这个答案中指出，自 2.7 版本以来，psycopg2 提供了一种或多或少安全的方法来做到这一点。

from psycopg2 import sql

sql_data_sample = """select * from {0}
           where dt = %(date_from)s
           and target in ('ACTIVE')
           ----------------------------------------------------
           union all
           ----------------------------------------------------
           (select * from {0}
           where dt = %(date_to)s
           and target in (%(class_target)s));"""

cur.execute(sql.SQL(sql_data_sample)
                .format(sql.Identifier(query_params['table_name'])), 
            query_params)

您可能需要从您的字典中删除

table_name

，我不确定 psycopg2 对参数字典中的其他项目有何反应，我现在无法测试它。

需要指出的是，这仍然存在 SQL 注入的风险，除非绝对必要，否则应该避免。通常，表名和字段名是查询字符串中相当固定的部分。

这是sql

模块

的相关文档。

Answer 2

您可以使用以下代码来消除缩进错误

sql_data_sample = str("""
select * from %s
where dt = %s
and target in ('ACTIVE')
----------------------------------------------------
union all
----------------------------------------------------
(select * from %s
where dt = %s
and target in (%s));""" %(
    "'" + str(list(query_params.values())[0]) + "'" +
    "'" + list(query_params.values())[1] + "'" +
    "'" + list(query_params.values())[2] + "'" +
    "'" + list(query_params.values())[3] + "'"
))

但是你需要再传递一个参数，因为你使用了 %s 5 次，但参数只有 4 个