如何在Python中安全地使用变量执行PostgreSQL查询

import psycopg2

inputed_email = "test@example.com" # 假设这是从用户输入获取的变量

conn = None
cur = None
try:
    conn = psycopg2.connect("dbname=postgres user=postgres password=postgres")
    cur = conn.cursor()

    # 错误的用法：直接将变量作为execute的独立参数，或进行字符串拼接
    # cur.execute("SELECT password FROM user WHERE email = ", inputed_email, ";")
    # 这种方式会导致 TypeError: function takes at most 2 arguments (3 given)
    # 即使使用字符串拼接，如 f"SELECT password FROM user WHERE email = '{inputed_email}';"
    # 也存在SQL注入风险且易出错

    print("尝试执行查询...")
    # 假设这里是错误的代码，为了演示问题，我们不会运行它
    # cur.execute("SELECT password FROM user WHERE email = ",inputed_email,";")
    # print(cur.fetchone())

except TypeError as e:
    print(f"捕获到错误: {e}")
    print("错误提示：execute() 函数最多接受两个参数，但您提供了三个。")
except Exception as e:
    print(f"发生其他错误: {e}")
finally:
    if cur:
        cur.close()
    if conn:
        conn.close()
    print("数据库连接已关闭。")

import psycopg2

inputed_email = "test@example.com" # 假设这是从用户输入获取的变量

conn = None
cur = None
try:
    conn = psycopg2.connect("dbname=postgres user=postgres password=postgres")
    cur = conn.cursor()

    # 正确的用法：使用 %s 占位符，并将变量作为 execute() 的第二个参数（一个列表或元组）
    sql_query = "SELECT password FROM public.user WHERE email = %s" # 注意：这里移除了末尾的分号，通常不是必需的
    cur.execute(sql_query, [inputed_email]) # 第二个参数必须是可迭代对象（如列表或元组）

    result = cur.fetchone()
    if result:
        print(f"找到用户密码: {result[0]}")
    else:
        print(f"未找到邮箱为 '{inputed_email}' 的用户。")

    conn.commit() # 对于SELECT语句通常不需要commit，但对于INSERT/UPDATE/DELETE是必需的

except psycopg2.Error as e:
    print(f"数据库操作错误: {e}")
    if conn:
        conn.rollback() # 发生错误时回滚事务
except Exception as e:
    print(f"发生其他错误: {e}")
finally:
    if cur:
        cur.close()
    if conn:
        conn.close()
    print("数据库连接已关闭。")

import psycopg2

user_email = "jane.doe@example.com"
user_lastname = "Doe"

conn = None
cur = None
try:
    conn = psycopg2.connect("dbname=postgres user=postgres password=postgres")
    cur = conn.cursor()

    # 多个变量的参数化查询
    sql_query = "SELECT firstname, password FROM public.user WHERE email = %s AND lastname = %s"
    cur.execute(sql_query, (user_email, user_lastname)) # 可以使用元组或列表

    result = cur.fetchone()
    if result:
        print(f"找到用户: {result[0]}, 密码: {result[1]}")
    else:
        print(f"未找到邮箱为 '{user_email}' 且姓氏为 '{user_lastname}' 的用户。")

    conn.commit()

except psycopg2.Error as e:
    print(f"数据库操作错误: {e}")
    if conn:
        conn.rollback()
except Exception as e:
    print(f"发生其他错误: {e}")
finally:
    if cur:
        cur.close()
    if conn:
        conn.close()
    print("数据库连接已关闭。")