基于SQLAlchemy实现JSON批量导入时的新增或更新功能

我来帮你梳理下这个问题，结合你的场景给出具体的解决方案：

先直接解答你的三个疑问 #

• 怎么高效实现批量新增/更新？
  不需要循环单个添加并提交，SQLAlchemy提供了原生的批量upsert（更新或插入）语法，针对不同数据库有对应的实现，一次请求就能完成所有数据的处理，效率比循环单条操作高得多。
• 是否需要把唯一字段改成键字段？
  完全不需要！你当前的模型设计没问题——id作为自增主键，ProductCode加unique=True作为唯一约束，这个组合既保留了自增主键的便利性（比如关联其他表时更高效），又能通过ProductCode判断数据是否重复，完美适配你的需求。
• 能不能保留自增键字段？
  当然可以！自增主键id会由数据库自动生成，不会干扰基于ProductCode的冲突判断逻辑，两者可以和谐共存。

问题出在哪？ #

你当前的代码有两个核心问题：

1. 循环里每次添加数据后就创建插入语句，还通过session.add逐条提交，这完全没用到批量操作的优势，效率极低；
2. 没有正确配置on_duplicate_key_update的更新字段，而且你的product_list里似乎漏掉了ProductCode——这可是判断数据重复的关键字段，必须包含进去！

具体实现代码 #

根据你使用的数据库类型，分两种情况给出代码：

如果你用的是MySQL

from sqlalchemy import insert

# 第一步：正确整理批量数据，必须包含ProductCode（唯一约束字段）
product_list = []
for product in products:
    product_data = {
        'ProductCode': product.get('ProductCode'),  # 这个是判断重复的核心，不能少
        'DisplayName': product.get('DisplayName'),
        'Description': product.get('Description'),
        # 其他需要同步的字段...
    }
    product_list.append(product_data)

# 第二步：构建批量插入语句
insert_stmt = insert(ProductDescriptor).values(product_list)

# 第三步：配置冲突时的更新逻辑——当ProductCode重复时，更新指定字段
on_duplicate_stmt = insert_stmt.on_duplicate_key_update(
    DisplayName=insert_stmt.inserted.DisplayName,
    Description=insert_stmt.inserted.Description,
    # 把需要更新的字段都列在这里，不想更新的就不要写
)

# 第四步：执行批量操作，一次搞定
try:
    self.session.execute(on_duplicate_stmt)
    self.session.commit()
except Exception as e:
    self.session.rollback()
    print(f"批量同步失败：{str(e)}")

如果你用的是PostgreSQL

PostgreSQL用的是on_conflict_do_update语法，需要明确指定冲突的约束字段：

from sqlalchemy import insert

# 同样先整理包含ProductCode的批量数据
product_list = []
for product in products:
    product_data = {
        'ProductCode': product.get('ProductCode'),
        'DisplayName': product.get('DisplayName'),
        'Description': product.get('Description'),
        # 其他字段...
    }
    product_list.append(product_data)

insert_stmt = insert(ProductDescriptor).values(product_list)

# 指定冲突的索引/约束（这里就是ProductCode的唯一约束），然后设置更新字段
on_conflict_stmt = insert_stmt.on_conflict_do_update(
    index_elements=['ProductCode'],  # 对应模型里的唯一字段名
    set_={
        'DisplayName': insert_stmt.inserted.DisplayName,
        'Description': insert_stmt.inserted.Description,
        # 其他需要更新的字段...
    }
)

# 执行操作
try:
    self.session.execute(on_conflict_stmt)
    self.session.commit()
except Exception as e:
    self.session.rollback()
    print(f"批量同步失败：{str(e)}")

补充说明 #

• insert_stmt.inserted.xxx表示：当发生冲突时，用本次插入请求中的新值覆盖数据库里的旧值，如果你有某些字段不想被更新（比如创建时间），就不要把它加到更新列表里；
• 这个批量操作的逻辑和你熟悉的Djangoupdate_or_create类似，但效率更高——Django的方法是单条处理，而SQLAlchemy的批量upsert只需要一次数据库交互，适合你这种JSON批量导入的场景。

备注：内容来源于stack exchange，提问作者Byte Insight