PostgreSQL批量UPSERT时Python端更新字段加密问题求助

我用Python的CryptoDome库加密字段，同时执行PostgreSQL批量UPSERT操作向新表插入数据，现有代码如下：

def bulk_upsert_query(data, table_name=None):
    values_list = []
    add = values_list.append

    for i in data:
        encrypt_user_id = encryption_fnc(str(i.user_id))

        add(f"({i.id},'{encrypt_user_id}','{i.name}')")

    upsert_sql = r"""insert into upsert_test_table 
                    (id,encrypt_user_id,name)
                   values
                   """ + "\n, ".join(values_list) + """ ON CONFLICT ON CONSTRAINT 
                  "upsert_test_table_pkey"
                   DO UPDATE SET
                   name=excluded.name,
                   encrypt_user_id=excluded.user_id"""
    return upsert_sql

• data来自旧表，user_id为整数类型（如124、345）
• upsert_test_table的id字段为主键

当前问题：插入时user_id加密逻辑正常，但主键冲突执行更新时，encrypt_user_id会被更新为未加密的原始user_id值。

需求：主键冲突更新时，encrypt_user_id必须调用Python端的encryption_fnc加密，禁止使用数据库端的pgcrypto；且user_id字段可变更，需保证更新时用最新的加密值。

问题根源是DO UPDATE部分错误引用了excluded.user_id——excluded代表INSERT语句中待插入的行数据，而你只插入了id、encrypt_user_id、name三个字段，excluded.user_id实际是取了未加密的原始值，正确的做法是直接引用excluded中已经加密好的encrypt_user_id字段。

修改后的代码 #

def bulk_upsert_query(data, table_name=None):
    values_list = []
    add = values_list.append

    for item in data:
        encrypt_user_id = encryption_fnc(str(item.user_id))
        # 插入时依然用加密后的值
        add(f"({item.id},'{encrypt_user_id}','{item.name}')")

    upsert_sql = r"""insert into upsert_test_table 
                    (id, encrypt_user_id, name)
                   values
                   """ + "\n, ".join(values_list) + """ 
                   ON CONFLICT ON CONSTRAINT "upsert_test_table_pkey"
                   DO UPDATE SET
                   name = excluded.name,
                   encrypt_user_id = excluded.encrypt_user_id"""
    return upsert_sql

关键改动说明 #

1. 修正字段引用：把DO UPDATE里的excluded.user_id改成excluded.encrypt_user_id。因为INSERT的VALUES中已经是Python端加密后的encrypt_user_id，excluded.encrypt_user_id直接就能拿到加密后的值，完全符合Python端加密的要求。
2. 可选优化：避免SQL注入：如果数据存在不可控内容，建议不要直接字符串拼接SQL，改用psycopg2的execute_values做参数化批量操作，更安全：

from psycopg2.extras import execute_values

def bulk_upsert(conn, data):
    records = []
    # 提前加密所有user_id
    for item in data:
        encrypted_user_id = encryption_fnc(str(item.user_id))
        records.append((item.id, encrypted_user_id, item.name))
    
    sql = """insert into upsert_test_table (id, encrypt_user_id, name)
             values %s
             ON CONFLICT ON CONSTRAINT "upsert_test_table_pkey"
             DO UPDATE SET
             name = excluded.name,
             encrypt_user_id = excluded.encrypt_user_id"""
    execute_values(conn.cursor(), sql, records)
    conn.commit()

这个方案不管是插入还是更新，encrypt_user_id都是Python端提前加密好的值，既满足了加密逻辑在Python端完成的要求，又能保证user_id变更时更新的是最新的加密结果。

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