使用SQLAlchemy和PostgreSQL批量导入时，已存在则更新属性

要实现当article字段存在时更新指定属性，不存在则插入的需求，我们可以利用PostgreSQL的ON CONFLICT语法（即Upsert），配合SQLAlchemy的异步API来高效完成批量操作，替代原有的bulk_save_objects方法。

修改后的批量导入函数 #

from sqlalchemy import insert

async def import(self, db: AsyncSession, objects):
    # 构造批量插入语句，提取每个Product实例的字段值
    insert_stmt = insert(Product).values(
        [
            {
                "idf": obj.idf,
                "name": obj.name,
                "article": obj.article,
                "category": obj.category,
                "description": obj.description
            }
            for obj in objects
        ]
    )
    
    # 自动筛选需要更新的字段（排除主键idf和唯一字段article）
    update_columns = {
        col.name: insert_stmt.excluded[col.name]
        for col in Product.__table__.columns
        if col.name not in ("idf", "article")
    }
    
    # 定义冲突处理逻辑：当article唯一约束冲突时，更新指定字段
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["article"],  # 基于article字段的唯一约束检测冲突
        set_=update_columns
    )
    
    # 执行Upsert语句并提交事务
    await db.execute(upsert_stmt)
    await db.commit()
    return {"status": "Import has been finished"}

关键逻辑说明 #

1. 构造批量插入语句：将传入的Product实例转换为字典列表，作为插入语句的数据源。
2. 自动筛选更新字段：通过遍历模型表的列，排除idf（主键）和article（唯一约束字段），自动生成需要更新的字段映射，后续新增字段时无需手动修改代码。
3. ON CONFLICT 处理：
   • index_elements=["article"]：指定以article字段的唯一约束作为冲突检测条件。
   • set_=update_columns：冲突发生时，用待插入数据（insert_stmt.excluded）中的值更新对应字段。
4. 异步执行：直接使用AsyncSession的execute方法执行Upsert语句，比原有的run_sync更适配异步场景，效率更高。

优势 #

• 性能高效：通过单条SQL语句完成批量插入/更新，避免了先查询再修改的N+1问题，适合大量数据导入。
• 原子性：整个Upsert操作是原子的，不会出现部分数据插入成功、部分失败的情况。
• 可扩展性：自动筛选更新字段的逻辑，让后续模型字段变更时无需修改Upsert代码。

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