如何使用psycopg2.sql避免psycopg2中使用f字符串与字符串拼接

实现代码 #

首先导入依赖模块：

import psycopg2
from psycopg2 import sql
from psycopg2.extras import execute_values

修改后的函数逻辑：

def insert_pairs(dat, no_transform_cols):
    # dat 是存储待插入数据的pandas dataframe
    no_transform_cols = list(no_transform_cols) # 不需要额外处理、可直接插入的字段列表
    
    cols2 = no_transform_cols + ['foo_id']                            
    cols = list(dat.columns) # 包含用于关联foo表的a字段的全量字段列表
    tuples = [tuple(x) for x in dat.to_numpy()] # 转换为元组用于批量插入
    
    # 构造动态SQL片段
    # 1. CTE临时表ins的字段列表
    cte_cols = sql.SQL(', ').join([sql.Identifier(col) for col in cols])
    # 2. 目标表bar的插入字段列表
    insert_cols = sql.SQL(', ').join([sql.Identifier(col) for col in cols2])
    # 3. SELECT子句中从ins表读取的字段列表
    select_ins_cols = sql.SQL(', ').join([sql.Identifier('ins', col) for col in no_transform_cols])
    
    # 拼接完整SQL语句
    query = sql.SQL('''
        WITH ins ({cte_cols}) AS (VALUES %s)
        INSERT INTO bar ({insert_cols})
        SELECT {select_ins_cols}, foo.id
        FROM ins
        LEFT JOIN foo ON foo.a = ins.a
    ''').format(
        cte_cols=cte_cols,
        insert_cols=insert_cols,
        select_ins_cols=select_ins_cols
    )
    
    # 执行查询示例（conn为已初始化的数据库连接）
    # cursor = conn.cursor()
    # execute_values(cursor, query, tuples)
    # conn.commit()

修改说明 #

• 所有动态传入的列名、表名均使用sql.Identifier包装，psycopg2会自动处理标识符转义，完全避免SQL注入风险
• 多个动态标识符之间的分隔拼接使用sql.SQL(', ').join()实现，替代原生字符串拼接逻辑
• 整体SQL模板通过sql.SQL定义，使用format方法传入动态构造的标识符片段，彻底替换f字符串拼接逻辑
• 批量值插入的%s是execute_values的专用占位符，不需要额外调整

内容的提问来源于stack exchange，提问作者sev