如何用Python编写通用函数实现表间记录的插入更新（不存在则插存在则更）

你的代码存在两个核心问题：

1. UPDATE语句的SET子句语法错误：直接写table_A = table_B是无效的，数据库无法识别这种整表赋值的写法，必须明确指定每一列的更新映射关系。
2. 非原子操作的竞态风险：分开执行INSERT和UPDATE，两次操作之间如果有其他进程修改表A的数据，会导致最终结果不符合预期。

推荐方案：使用数据库原生UPSERT语法（原子操作） #

UPSERT是数据库原生的“插入或更新”原子操作，比分开执行INSERT/UPDATE更高效且可靠，不同数据库的语法略有差异，以下是两种主流数据库的实现：

PostgreSQL版本 #

def insert_update_record(table_A, table_B, pk_column='id'):
    # 获取表B的所有列（假设表A和表B结构完全一致）
    cur.execute(f"SELECT column_name FROM information_schema.columns WHERE table_name = %s", (table_B,))
    columns = [row[0] for row in cur.fetchall()]
    
    # 构造列列表和更新子句
    columns_str = ', '.join(columns)
    update_clause = ', '.join([f"{col} = EXCLUDED.{col}" for col in columns if col != pk_column])
    
    # 构造UPSERT语句
    upsert_query = f"""
        INSERT INTO {table_A} ({columns_str})
        SELECT {columns_str} FROM {table_B}
        ON CONFLICT ({pk_column}) DO UPDATE SET {update_clause}
    """
    
    cur.execute(upsert_query)
    conn.commit()

MySQL版本 #

def insert_update_record(table_A, table_B, pk_column='id'):
    # 获取表B的所有列（假设表A和表B结构完全一致）
    cur.execute(f"SELECT column_name FROM information_schema.columns WHERE table_name = %s", (table_B,))
    columns = [row[0] for row in cur.fetchall()]
    
    # 构造列列表和更新子句
    columns_str = ', '.join(columns)
    update_clause = ', '.join([f"{col} = VALUES({col})" for col in columns if col != pk_column])
    
    # 构造UPSERT语句
    upsert_query = f"""
        INSERT INTO {table_A} ({columns_str})
        SELECT {columns_str} FROM {table_B}
        ON DUPLICATE KEY UPDATE {update_clause}
    """
    
    cur.execute(upsert_query)
    conn.commit()

修复原有INSERT/UPDATE方案（不推荐） #

如果必须分开执行插入和更新，需要修正UPDATE语句的SET子句，同时确保表A和表B的列名完全一致：

def insert_update_record(table_A, table_B):
    # 插入表A中不存在的记录（修正关联逻辑的顺序，更易读）
    insert_query = """
        INSERT INTO {} 
        SELECT * FROM {} 
        WHERE NOT EXISTS (
            SELECT 1 FROM {} 
            WHERE {}.id = {}.id
        )
    """.format(table_A, table_B, table_A, table_A, table_B)
    
    # 获取表A的列名，构造更新子句
    cur.execute(f"SELECT column_name FROM information_schema.columns WHERE table_name = %s", (table_A,))
    columns = [row[0] for row in cur.fetchall()]
    set_clause = ', '.join([f"{table_A}.{col} = {table_B}.{col}" for col in columns if col != 'id'])
    
    # 修正后的UPDATE语句
    update_query = """
        UPDATE {} 
        SET {} 
        FROM {} 
        WHERE {}.id = {}.id
    """.format(table_A, set_clause, table_B, table_A, table_B)
    
    cur.execute(insert_query)
    cur.execute(update_query)
    conn.commit()

说明 #

• 该方案仍存在竞态风险，若业务对数据一致性要求高，优先使用UPSERT。
• 代码中通过查询information_schema.columns获取列名，实现了一定的通用性，但前提是表A和表B的结构完全匹配。

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