如何在Python中实现MSSQL数据的UPSERT操作

从API获取JSON格式数据，需加载至已有MSSQL表中。API返回数据格式如下：

{
  "abcd": "blablabla",
  "value": [
    {
        "a": "1",
        "b": "2",
        "c": "3"
    },
    {
        "a": "4",
        "b": "5",
        "c": "6"
    }
  ]
}

仅关注value字段数据，表列与a、b、c对应，已用以下代码转为pandas DataFrame：

contents = pandas.json_normalize(response.json(), record_path=['value'])

当前使用to_sql的append模式插入数据，但需要实现UPSERT：当a列（指定键）不存在时插入，存在时更新。尝试SQLAlchemy的merge方法但未成功，现有数据库连接及加载代码如下：

with open('db_con.json') as db:
            con = json.load(db)

            con_url = URL.create(
                "mssql+pyodbc",
                username = con['username'],
                password = con['password'],
                host = con['host'],
                port = con['port'],
                database = con['database'],
                query = con['query'],
                )
            engine = sqlalchemy.create_engine(con_url, fast_executemany = False, echo = True)

            contents.to_sql(
                test_db,
                engine,
                if_exists='append',
                index = False,
                chunksize = 2500,
                schema = 'test'
            )

请问如何实现UPSERT？是否还能使用to_sql？

方法1：临时表 + MERGE语句（推荐，性能最优） #

可以继续使用to_sql，先将数据写入临时表，再通过MSSQL原生的MERGE语句完成UPSERT，步骤如下：

1. 将DataFrame写入MSSQL临时表：

# 写入本地临时表（会话关闭后自动销毁）
contents.to_sql(
    '#temp_test',
    engine,
    if_exists='replace',
    index=False,
    chunksize=2500,
    schema='test'
)

2. 执行MERGE语句，以a为匹配键完成更新/插入：

merge_sql = """
MERGE INTO test.test_db AS target
USING #temp_test AS source
ON target.a = source.a
WHEN MATCHED THEN
    UPDATE SET
        b = source.b,
        c = source.c
WHEN NOT MATCHED THEN
    INSERT (a, b, c)
    VALUES (source.a, source.b, source.c);
"""

# 在同一连接会话中执行MERGE并提交
with engine.connect() as conn:
    conn.execute(sqlalchemy.text(merge_sql))
    conn.commit()

注意：

• 本地临时表（#前缀）仅在当前连接会话可见，会话关闭后自动删除，需确保写入临时表和执行MERGE在同一个连接中完成。
• 若需要跨会话使用临时表，可改用全局临时表（##前缀），但需注意并发冲突问题。
• 目标表的a列必须是主键或具有唯一约束，否则MERGE可能出现多行匹配的异常。

方法2：SQLAlchemy Core 构造MERGE操作 #

如果不想使用临时表，可通过SQLAlchemy Core直接构造MERGE语句处理数据，适合小数据量场景：

1. 映射目标表结构：

from sqlalchemy import MetaData, Table, Column, String

metadata = MetaData(schema='test')
target_table = Table(
    'test_db',
    metadata,
    Column('a', String, primary_key=True),  # 确保a是主键/唯一键
    Column('b', String),
    Column('c', String)
)

2. 遍历DataFrame执行MERGE：

from sqlalchemy.dialects.mssql import merge

with engine.connect() as conn:
    for _, row in contents.iterrows():
        # 构造MERGE语句
        merge_stmt = merge(target_table, values={
            'a': row['a'],
            'b': row['b'],
            'c': row['c']
        }).on(target_table.c.a == row['a'])
        
        # 匹配时更新字段
        merge_stmt = merge_stmt.when_matched_update(set_={
            'b': row['b'],
            'c': row['c']
        })
        
        # 不匹配时插入新行
        merge_stmt = merge_stmt.when_not_matched_insert(values={
            'a': row['a'],
            'b': row['b'],
            'c': row['c']
        })
        
        conn.execute(merge_stmt)
    conn.commit()

说明：该方法逐行处理数据，数据量大时效率较低，优先选择方法1。

总结 #

• 完全可以继续使用to_sql，临时表+MERGE的方案是批量UPSERT的最优选择，兼顾效率和实现复杂度。
• SQLAlchemy的merge方法需要结合表元数据映射使用，适合小规模数据场景。
• 核心前提：目标表的a列必须具备唯一性约束（主键或唯一索引），否则UPSERT逻辑无法正确执行。

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