如何用Python脚本构造支持多类型的PostgreSQL跨表插入SQL语句

问题核心 #

手动拼接INSERT语句时，未正确处理全量数据类型（如浮点数、字符串的特殊格式要求），且存在SQL注入风险、效率低下等问题。psycopg2本身提供了更安全高效的方案来处理这类数据复制场景。

方案1：使用executemany批量插入（推荐） #

利用psycopg2的参数化查询机制，无需手动拼接值，驱动会自动处理类型转换与SQL注入防护：

def copy_table_data(table, create_table_columns, cur_to, cur_from):
    table_name_from = table['table_name_from']
    table_name_to = table['table_name_to']
    table_schema = table['table_schema']
    
    # 获取源表数据
    sql_select = f"SELECT * FROM {table_schema}.{table_name_from}"
    cur_from.execute(sql_select)
    table_data = cur_from.fetchall()
    
    if not table_data:
        return  # 无数据直接返回
    
    # 构造INSERT语句的列与占位符部分
    column_names = [col['column_name'] for col in create_table_columns]
    columns_str = ", ".join(column_names)
    placeholders = ", ".join(["%s"] * len(column_names))
    sql_insert = f"INSERT INTO {table_schema}.{table_name_to} ({columns_str}) VALUES ({placeholders})"
    
    # 批量插入数据
    cur_to.executemany(sql_insert, table_data)
    # 提交操作可由调用方统一处理，或在此处添加 cur_to.connection.commit()

方案2：使用copy_from实现高效数据复制（大数据量首选） #

PostgreSQL的COPY命令是批量数据迁移的最优方式，psycopg2的copy_from方法直接支持该功能：

def copy_table_data_fast(table, cur_to, cur_from):
    table_name_from = table['table_name_from']
    table_name_to = table['table_name_to']
    table_schema = table['table_schema']
    
    # 用内存缓冲区中转数据
    from io import StringIO
    buffer = StringIO()
    
    # 导出源表数据到缓冲区
    cur_from.copy_to(buffer, f"{table_schema}.{table_name_from}", sep='\t', null='\\N')
    buffer.seek(0)
    
    # 从缓冲区导入到目标表
    cur_to.copy_from(buffer, f"{table_schema}.{table_name_to}", sep='\t', null='\\N')
    # 提交操作由调用方处理

方案3：手动构造INSERT语句（仅特殊场景使用） #

若必须手动拼接SQL，需针对每种类型做格式化处理，同时注意字符串转义：

def copy_table_statement(table, create_table_columns, cur_to, cur_from):
    table_name_from = table['table_name_from']
    table_name_to = table['table_name_to']
    table_schema = table['table_schema']
    sqlstring = f"SELECT * FROM {table_schema}.{table_name_from}"
    
    cur_from.execute(sqlstring)
    table_data = cur_from.fetchall()
    
    if not table_data:
        return ""
    
    # 构造列名部分
    column_names = [col['column_name'] for col in create_table_columns]
    columns_str = ", ".join(column_names)
    copy_table_sql_statement = f"INSERT INTO {table_schema}.{table_name_to} ({columns_str}) VALUES "
    
    # 逐行处理数据类型
    rows = []
    for row in table_data:
        processed_values = []
        for val in row:
            if val is None:
                processed_values.append("NULL")
            elif isinstance(val, bool):
                processed_values.append("TRUE" if val else "FALSE")
            elif isinstance(val, (int, float)):
                processed_values.append(str(val))
            elif isinstance(val, str):
                # 转义单引号避免SQL语法错误
                escaped_val = val.replace("'", "''")
                processed_values.append(f"'{escaped_val}'")
            # 可扩展处理日期、字节等其他类型
            else:
                escaped_val = str(val).replace("'", "''")
                processed_values.append(f"'{escaped_val}'")
        rows.append(f"({', '.join(processed_values)})")
    
    copy_table_sql_statement += ", ".join(rows) + ";"
    return copy_table_sql_statement

关键提示 #

• 方案1、2完全规避手动拼接值的风险，psycopg2会自动适配所有PostgreSQL数据类型，同时杜绝SQL注入。
• 方案3仅适合特殊场景，需严格处理各类数据格式，否则易出现语法错误或安全漏洞。

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