Se você deseja passar argumentos nomeados para
cursor.execute() , você pode usar o %(name)s sintaxe e passar um dict. Consulte a documentação
para mais detalhes. Aqui está um exemplo usando sua consulta:
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',)
)
