如何在SQL查询语句的占位位置传入Python列表执行
Python列表传入SQL占位符的正确实现
绝对不要直接把列表转成原始字符串用str.format()、f-string拼接进SQL语句,轻则碰到特殊字符(比如字符串里的单引号)触发SQL语法错误,重则导致SQL注入漏洞。
根据占位符的实际用途,分两种场景处理:
场景1:SELECT后占位符传入动态字段列表
你写的两个{}在SELECT关键字之后,属于查询字段(SQL标识符)位置,这类位置数据库驱动的参数化机制不支持传参,必须走「白名单校验+安全拼接」逻辑,同时要注意EXCEPT前后两个SELECT查询的字段数量、顺序、数据类型必须完全一致。
- 第一步:提前定义允许查询的字段白名单,从根源避免SQL注入
- 第二步:校验两个待传入字段列表的所有值都在白名单范围内
- 第三步:将校验通过的列表转为逗号分隔的字段字符串
- 第四步:将字段字符串拼入SQL模板生成最终可执行语句
实现代码示例(以PostgreSQL的psycopg2驱动为例):
# 两个待传入的字段列表,必须保证字段数量、类型顺序匹配 select_fields_1 = ["id", "desc_id", "site_name", "jump_url"] select_fields_2 = ["id", "desc_id", "site_name", "jump_url"] # 提前定义允许查询的字段白名单 allowed_fields = {"id", "desc_id", "site_name", "jump_url", "create_time", "is_valid"} # 校验字段合法性 for field in select_fields_1 + select_fields_2: if field not in allowed_fields: raise ValueError(f"存在非法查询字段: {field}") # 列表转逗号分隔字符串 fields_str_1 = ", ".join(select_fields_1) fields_str_2 = ", ".join(select_fields_2) # 拼接生成最终SQL final_sql = f"""select {fields_str_1} from public."myTempTable" where desc_id in (select desc_id from projectapp_description_table) Except select {fields_str_2} from projectapp_sitelink_table"""
场景2:IN子句传入值列表
如果你实际是要给where desc_id in (...)的位置传入Python值列表,替换掉原有子查询,直接用数据库驱动自带的参数化传参能力即可,不需要手动拼接值:
PostgreSQL(psycopg2)写法
psycopg2原生支持列表适配,用ANY(%s)语法即可直接传入列表:
# 待传入的desc_id匹配列表 desc_id_list = [2, 4, 6, 8, 10] # 字段部分按上面的白名单逻辑提前拼接完成 final_sql = """select {fields1} from public."myTempTable" where desc_id = ANY(%s) Except select {fields2} from projectapp_sitelink_table""".format(fields1=fields_str_1, fields2=fields_str_2) # 执行时直接把列表作为参数传入,驱动会自动做类型转义、格式适配 cursor.execute(final_sql, (desc_id_list,)) query_result = cursor.fetchall()
MySQL(pymysql)/ SQLite写法
这两个驱动不支持直接传列表到IN子句,可以先按列表长度生成对应数量的参数占位符,再把列表拆成参数传入:
desc_id_list = [2,4,6,8,10] # 生成和列表长度一致的占位符串,比如长度为5就生成 "%s,%s,%s,%s,%s" placeholder_str = ", ".join(["%s"] * len(desc_id_list)) final_sql = """select {fields1} from public."myTempTable" where desc_id in ({placeholder}) Except select {fields2} from projectapp_sitelink_table""".format( fields1=fields_str_1, fields2=fields_str_2, placeholder=placeholder_str ) # 把列表拆成位置参数传入 cursor.execute(final_sql, tuple(desc_id_list)) query_result = cursor.fetchall()
注意:所有值类型的参数都不要手动拼接进SQL字符串,必须通过驱动的
execute方法的第二个参数传入,才能从根本上避免SQL注入和语法错误。
内容的提问来源于stack exchange,提问作者Aashish
相关产品推荐
相关产品推荐

