基于SQLAlchemy/Pandas实现SQL数据库Upsert/Append需求问询

核心方案：利用SQL Server MERGE语句实现批量Upsert #

针对你使用SQL Server（SSMS 19）的场景，直接用T-SQL的MERGE语句就能实现匹配主键时按规则更新、不匹配时插入的逻辑，无需依赖PostgreSQL专属工具，也不用逐行手动处理。

步骤1：将DataFrame写入临时表 #

先把待处理的DataFrame数据写入SQL Server本地临时表，批量处理更高效：

from sqlalchemy import text, BigInteger, String, Text, Date

# 复用你原有的列类型定义
dtype_dict = {
    'Priority': BigInteger,
    'PN': String(255),
    'Part_Type': Text,
    'MPN': Text,
    'Proposed_MPN': Text,
    'Distributor': String(100),
    'Reason_for_Add': Text,
    'Line_Down': Date,
    'Engineering-Samples_Available': Text,
    'Build_Samples-Available': Text,
    'Comments': Text,                                                                                       
    'Supplier_Comments': Text,
    'Circuit_Location': Text,
    'Last_Updated': Date 
}

# 写入本地临时表（会话结束后自动销毁）
temp_table = '#ShortageTemp'
fulldf.to_sql(temp_table, engine, if_exists='replace', index=False, dtype=dtype_dict)

步骤2：执行MERGE语句实现自定义Upsert #

编写MERGE语句，针对不同列设置对应的更新规则：

• 固定值列（Part_Type、MPN、Distributor等）：用新值覆盖（需保留旧值则改为Target.列名）
• 日期列（Line_Down、Last_Updated）：直接替换为新日期
• 文本列（Comments、Supplier_Comments）：将新内容与旧内容拼接，用换行分隔，自动处理空值

# 编写MERGE的T-SQL语句
merge_sql = """
MERGE INTO Shortage AS Target
USING #ShortageTemp AS Source
ON Target.PN = Source.PN  -- 按主键PN匹配
WHEN MATCHED THEN
    UPDATE SET
        -- 固定值列：用新值覆盖，要忽略则改为 Target.列名
        Priority = Source.Priority,
        Part_Type = Source.Part_Type,
        MPN = Source.MPN,
        Proposed_MPN = Source.Proposed_MPN,
        Distributor = Source.Distributor,
        Reason_for_Add = Source.Reason_for_Add,
        -- 日期列：替换为新日期
        Line_Down = Source.Line_Down,
        [Engineering-Samples_Available] = Source.[Engineering-Samples_Available],
        [Build_Samples-Available] = Source.[Build_Samples-Available],
        -- 文本列：拼接新内容+旧内容，用换行分隔，处理空值
        Comments = ISNULL(Source.Comments, '') + CASE WHEN Target.Comments IS NOT NULL THEN CHAR(13)+CHAR(10)+Target.Comments ELSE '' END,
        Supplier_Comments = ISNULL(Source.Supplier_Comments, '') + CASE WHEN Target.Supplier_Comments IS NOT NULL THEN CHAR(13)+CHAR(10)+Target.Supplier_Comments ELSE '' END,
        Circuit_Location = Source.Circuit_Location,
        Last_Updated = Source.Last_Updated
WHEN NOT MATCHED THEN
    -- 无匹配时插入全量数据
    INSERT (Priority, PN, Part_Type, MPN, Proposed_MPN, Distributor, Reason_for_Add, Line_Down, [Engineering-Samples_Available], [Build_Samples-Available], Comments, Supplier_Comments, Circuit_Location, Last_Updated)
    VALUES (Source.Priority, Source.PN, Source.Part_Type, Source.MPN, Source.Proposed_MPN, Source.Distributor, Source.Reason_for_Add, Source.Line_Down, Source.[Engineering-Samples_Available], Source.[Build_Samples-Available], Source.Comments, Source.Supplier_Comments, Source.Circuit_Location, Source.Last_Updated);
"""

# 执行SQL并提交事务
with engine.connect() as conn:
    conn.execute(text(merge_sql))
    conn.commit()

关键配置说明 #

1. 忽略特定列：如果某些列（比如Part_Type）不需要更新，只需把UPDATE中的对应行改成Part_Type = Target.Part_Type，即可保留数据库原有值。
2. 文本拼接顺序：若需将旧内容放在前面，调换Source和Target的位置即可，比如Comments = ISNULL(Target.Comments, '') + CHAR(13)+CHAR(10)+ISNULL(Source.Comments, '')。
3. 空值处理：用ISNULL函数把NULL转为空字符串，避免拼接结果为NULL。
4. 临时表安全：本地临时表（#开头）仅在当前连接会话有效，会话结束后自动删除，不会污染数据库。

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