Parameterized Queries with Psycopg2/Python Db-API and Postgresql

Parameterized queries with psycopg2 / Python DB-API and PostgreSQL

psycopg2 follows the rules for DB-API 2.0 (set down in PEP-249). That means you can call execute method from your cursor object and use the pyformat binding style, and it will do the escaping for you. For example, the following should be safe (and work):

cursor.execute("SELECT * FROM student WHERE last_name = %(lname)s", 
               {"lname": "Robert'); DROP TABLE students;--"})

make parameterized SQL select statement with psycopg2 in python

First this is a duplicate of your previous question Previous. You should have just continued the discussion there. As I stated there it is possible to do what you using the sql module from psycopg2. As example from one of my applications:

class NotificationReport():
    """Builds a query for finding task notifications.

    Use form_choices passed in to modify the select query for task
    notifications using psycopg2.sql module. Filter on status which is some
    combination of notify_expired and notify_cancelled.
    """

    def __init__(self, form_choices):
        self.id = "notification_report"
        self.form_choices = form_choices

    def returnQuery(self):
        flds, defaults = data.fetchFields(data.TaskNotification)
        base_sql = sql.SQL("""SELECT
            task_title, {}
        FROM
            tasks
        JOIN
            task_priority AS tp
        ON
            tasks. task_priority_fk= tp.priority_id
        JOIN
            task_type AS tt
        ON
            tasks.task_type_fk = tt.task_type_id
        LEFT JOIN
            task_notification AS tn
        ON
            tasks.task_id = tn.task_id_fk

        """).format(sql.SQL(",").join(map(sql.Identifier, flds)))
        f_choices = self.form_choices
        and_sql = None
        ops_list = []
        if f_choices:
            for choice in f_choices:
                if choice.get("status"):
                    status = choice["status"]
                    status_dict = {"open": ("notify_expired = 'f' "),
                                   "expired": ("notify_expired = 't' "),
                                   }
                    if status == "all":
                        pass
                    else:
                        ops = sql.SQL(status_dict[status])
                        ops_list.append(ops)
        if ops_list:
            and_sql = sql.Composed([base_sql, sql.SQL(" AND ")])
            additional_and = sql.SQL(" AND ").join(ops_list)
            ops_sql = sql.Composed([and_sql, additional_and])
        orderby_sql = sql.SQL("""ORDER BY
            task_title""")
        if and_sql:
            combined_sql = sql.Composed([ops_sql, orderby_sql])
        else:
            combined_sql = sql.Composed([base_sql, orderby_sql])

        return combined_sql

Output. First no parameters supplied to report:

SELECT
    task_title, "task_id_fk","before_value","before_interval","every_value","every_interval","notify_note","notify_id","notify_expired"
FROM
    tasks
JOIN
    task_priority AS tp
ON
    tasks. task_priority_fk= tp.priority_id
JOIN
    task_type AS tt
ON
    tasks.task_type_fk = tt.task_type_id
LEFT JOIN
    task_notification AS tn
ON
    tasks.task_id = tn.task_id_fk

ORDER BY
    task_title

Then with status:

SELECT
    task_title, "task_id_fk","before_value","before_interval","every_value","every_interval","notify_note","notify_id","notify_expired"
FROM
    tasks
JOIN
    task_priority AS tp
ON
    tasks. task_priority_fk= tp.priority_id
JOIN
    task_type AS tt
ON
    tasks.task_type_fk = tt.task_type_id
LEFT JOIN
    task_notification AS tn
ON
    tasks.task_id = tn.task_id_fk

    AND notify_expired = 'f' ORDER BY
    task_title

Psycopg2 parameterized execute query

You might want to use a multiline string to define your SQL query, to provide more meaningful names you can use a dictionary to pass values to psycopg2.execute():

import psycopg2
conn = psycopg2.connect("dbname=mf port=5959 host=localhost user=mf_usr")
cur = conn.cursor()

sql = """
    SELECT
        id
    FROM tbl_users
    WHERE
        b_enabled = %(enabled)s
"""

print (cur.mogrify(sql, {'enabled': "True"}).decode('utf-8'))
# cur.execute(sql, {'enabled': "True"})

Output:

SELECT
    id
FROM tbl_users
WHERE
    b_enabled = 'True'

Please have a look at the official docs for further information.

any way to make parameterized queries and encapsulate it python in function

If you want to pass named arguments to cursor.execute(), you can use the %(name)s syntax and pass in a dict. See the documentation for more details.

Here is an example using your query:

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',)
    )

id variable in postgres

In Python's Postgres library, you can interpolate values with %s and pass them as secondary arguments to execute.

my_cursor.execute("SELECT word FROM words WHERE id = %s", (my_id,))

Passing list of parameters to SQL in psycopg2

Python tuples are converted to sql lists in psycopg2:

cur.mogrify("SELECT * FROM table WHERE column IN %s;", ((1,2,3),))

would output

'SELECT * FROM table WHERE column IN (1,2,3);'

For Python newcomers: It is unfortunately important to use a tuple, not a list here. Here's a second example:

cur.mogrify("SELECT * FROM table WHERE column IN %s;", 
    tuple([row[0] for row in rows]))

Unable to Pass in Table Name for Query psycopg2

According to:

Passing table name as a parameter in psycopg2

You should be using this

template:

from psycopg2 import sql
cur.execute(
    sql.SQL("insert into {table} values (%s, %s)")
        .format(table=sql.Identifier('my_table')), # table name here
    [10, 20]) ## other parameters here

---

Related Topics