python - 进行参数化查询并将其封装在python函数中的任何方法

Question

我想实现一个 python 函数，它将执行带参数的 SQL 查询。为此，我开始使用 psycopg2 来访问我的本地数据库。然而，我写了一堆非常相似的 SQL 查询，而每个 SQL 语句在取不同值方面都略有不同。我的目标是我想编写参数化 SQL，以便我可以将其包装在 python 函数中，理想情况下，我可以使用任意参数进行函数调用，以便它可以替换 SQL 语句中的参数值。我查看了 SO 帖子并得到了一些想法，但无法完成可以使用任意参数执行 SQL 语句的紧凑型 python 函数。我知道如何通过使用 **kwargs、*args 来编写传递任意参数的 python 函数，但不确定如何在 python 函数中对参数化 SQl 执行此操作。有没有什么有效的方法可以轻松地在 python 中执行此操作？有什么可行的方法可以实现这一点？

我的数据库架构:

这是我在 postgresql 中的表模式:

CREATE TABLE mytable(
date_received DATE,
pk_est VARCHAR,
grd_name VARCHAR,
cl_val SMALLINT,
quant_received NUMERIC,
mg_fb_price NUMERIC,
freight NUMERIC,
standard_price NUMERIC,
grd_delv_cost NUMERIC,
order_type VARCHAR,
pk_name VARCHAR,
item_type VARCHAR,
waiting_days NUMERIC,
item_name VARCHAR,
mk_price_variance NUMERIC,
);

我的示例 SQL 查询

这是需要参数化的示例 SQL 查询之一:

SELECT
    date_trunc('week', date_received) AS received_week,
    cl_val,
    ROUND(ROUND(SUM(quant_received * standard_price)::numeric,4) / SUM(quant_received),4) AS mk_price_1,
    ROUND(ROUND(SUM(quant_received * mg_fb_price)::numeric,4) / SUM(quant_received),4) AS mg_price_1,
    ROUND(ROUND(SUM(quant_received * mk_price_variance)::numeric,4) / SUM(quant_received),4) AS fb_mk_price_var,
    ROUND(ROUND(SUM(quant_received * freight)::numeric,4) / SUM(quant_received),4) AS freight_new,
    ROUND(ROUND(SUM(quant_received * grd_delv_cost)::numeric,4) / SUM(quant_received),4) AS grd_delv_cost_new,
    TO_CHAR(SUM(quant_received), '999G999G990D') AS Volume_Received
FROM mytable
WHERE date_received >= to_date('2010-10-01','YYYY-MM-DD')
    AND date_received <= to_date('2012-12-31','YYYY-MM-DD')
    AND item_type = 'processed'
    AND cl_val IN ('12.5','6.5','8.1','8.5','9.0')
    AND order_type IN ('formula')
    AND pk_name IN ('target','costco','AFG','KFC')
    AND pk_est NOT IN ('12')
GROUP BY received_week,cl_val
ORDER BY received_week ASC ,cl_val ASC;

我目前的尝试:

import psycopg2

connection = psycopg2.connect(database="myDB", user="postgres", password="passw", host="localhost", port=5432)
cursor = connection.cursor()
cursor.execute(
    """
    select * from mytable where date_received < any(array['2019-01-01'::timestamp, '2020-07-10'::timestamp])
    """)
record = cursor.fetchmany()

另一个尝试:

cursor.execute("""
    select date_trunc('week', date_received) AS received_week,
        cl_val,
        ROUND(ROUND(SUM(quant_received * standard_price)::numeric,4) / SUM(quant_received),4) AS mk_price_1,
    from (
        select * from mytable 
        where  item_type = %s and order_type IN %s
    ) t;
""", (item_type_value, order_type_value))

results = [r[0] for r in cursor.fetchall()]

但是在我的代码中，有很多硬编码的部分需要参数化。我想知道在 python 中是否有任何方法可以做到这一点。谁能指出我如何实现这一目标？在 python 函数中实现参数化 SQL 是否可行？任何想法？谢谢

目标

我希望实现这样的功能:

def parameterized_sql(**kwargs, *args):
    connection = psycopg2.connect(database="myDB", user="postgres", password="passw", host="localhost", port=5432)
    cursor = connection.cursor()
    cursor.execute("""SQL statement with parameter""")
    ## maybe more

这只是 python 函数的框架，我想实现它但不确定它是否可行。任何反馈都会有所帮助。谢谢

更新:

我期待可以将参数值传递给 SQL 主体的通用 python 函数，这样我就可以避免编写许多 SQL 查询，这些查询实际上彼此之间有很多重叠，并且没有参数化。目标是制作可在 python 函数中执行的参数化 SQL 查询。

Answer 1

如果你想将命名参数传递给 cursor.execute()，你可以使用 %(name)s 语法并传入一个字典。参见 the documentation了解更多详情。

这是使用您的查询的示例:

import datetime
import psycopg2

EXAMPLE_QUERY = """
SELECT
    date_trunc('week', date_received) AS received_week,
    cl_val,
    ROUND(ROUND(SUM(quant_received * standard_price)::numeric,4) / SUM(quant_received),4) AS mk_price_1,
    ROUND(ROUND(SUM(quant_received * mg_fb_price)::numeric,4) / SUM(quant_received),4) AS mg_price_1,
    ROUND(ROUND(SUM(quant_received * mk_price_variance)::numeric,4) / SUM(quant_received),4) AS fb_mk_price_var,
    ROUND(ROUND(SUM(quant_received * freight)::numeric,4) / SUM(quant_received),4) AS freight_new,
    ROUND(ROUND(SUM(quant_received * grd_delv_cost)::numeric,4) / SUM(quant_received),4) AS grd_delv_cost_new,
    TO_CHAR(SUM(quant_received), '999G999G990D') AS Volume_Received
FROM mytable
WHERE date_received >= %(min_date_received)s
    AND date_received <= %(max_date_received)s
    AND item_type = %(item_type)s
    AND cl_val IN %(cl_vals)s
    AND order_type IN %(order_types)s
    AND pk_name IN %(pk_names)s
    AND pk_est NOT IN %(pk_ests)s
GROUP BY received_week ,cl_val
ORDER BY received_week ASC, cl_val ASC;
"""


def execute_example_query(cursor, **kwargs):
    """Execute the example query with given parameters."""
    cursor.execute(EXAMPLE_QUERY, kwargs)
    return cursor.fetchall()
            

if __name__ == '__main__':
    connection = psycopg2.connect(database="myDB", user="postgres", password="passw", host="localhost", port=5432)
    cursor = connection.cursor()
    execute_example_query(
        cursor,
        min_date_received = datetime.date(2010, 10, 1),
        max_date_received = datetime.date(2012, 12, 31),
        item_type = 'processed',
        cl_vals = ('12.5', '6.5', '8.1', '8.5', '9.0'),
        order_types = ('formula',),
        pk_names = ('target', 'costco', 'AFG', 'KFC'),
        pk_ests = ('12',)
    )