SQLAlchemy ORM中session.add触发重复键错误时如何实现更新而非插入

核心原因：你之前使用merge()方法不生效的原因是，SQLAlchemy的merge()默认仅通过主键匹配已存在的记录执行更新逻辑，你的唯一约束建立在stock_symbol字段上（非表主键），因此merge()无法识别到重复记录，仍然会尝试执行插入操作触发冲突。

方案1：先查询后写入（全版本兼容） #

该方案兼容所有SQLAlchemy版本和数据库类型，逻辑直观无额外依赖：

for row in symbols:
    # 先查询是否存在对应stock_symbol的记录
    exist_stock = self.session.query(Trial_table).filter_by(stock_symbol=row["symbol"]).first()
    if exist_stock:
        # 存在则直接更新字段
        exist_stock.company_name = row["companyName"]
        exist_stock.last_updated = arrow.utcnow().to('local').format('YYYY-MM-DD HH:mm:ss')
    else:
        # 不存在则新建记录
        stock_table = Trial_table()
        stock_table.stock_symbol = row["symbol"]
        stock_table.company_name = row["companyName"]
        stock_table.last_updated = arrow.utcnow().to('local').format('YYYY-MM-DD HH:mm:ss')
        self.session.add(stock_table)

# 循环结束后统一commit，避免频繁IO
try:
    self.session.commit()
except:
    self.session.rollback()
    raise

方案2：Upsert原子操作（性能最优推荐） #

如果你使用SQLAlchemy 1.4+、MySQL 5.7+，可以直接用数据库原生的冲突更新语法，是原子操作，无并发问题，性能远高于先查后写：

from sqlalchemy.dialects.mysql import insert

for row in symbols:
    insert_stmt = insert(Trial_table).values(
        stock_symbol = row["symbol"],
        company_name = row["companyName"],
        last_updated = arrow.utcnow().to('local').format('YYYY-MM-DD HH:mm:ss')
    )
    # 冲突时更新除唯一键外的其他字段
    on_conflict_stmt = insert_stmt.on_duplicate_key_update(
        company_name = insert_stmt.inserted.company_name,
        last_updated = insert_stmt.inserted.last_updated
    )
    self.session.execute(on_conflict_stmt)

try:
    self.session.commit()
except:
    self.session.rollback()
    raise

方案3：冲突捕获后更新（兼容原有逻辑） #

如果不想修改原有写入逻辑，也可以在捕获到唯一键冲突时执行更新操作：

from sqlalchemy.exc import IntegrityError

for row in symbols:
    stock_table = Trial_table()
    stock_table.stock_symbol = row["symbol"]
    stock_table.company_name = row["companyName"]
    stock_table.last_updated = arrow.utcnow().to('local').format('YYYY-MM-DD HH:mm:ss')
    self.session.add(stock_table)
    try:
        self.session.commit()
    except IntegrityError as e:
        self.session.rollback()
        # 唯一键冲突时执行更新
        self.session.query(Trial_table).filter_by(stock_symbol=row["symbol"]).update({
            "company_name": row["companyName"],
            "last_updated": arrow.utcnow().to('local').format('YYYY-MM-DD HH:mm:ss')
        })
        self.session.commit()
    except:
        self.session.rollback()
        raise

优化建议 #

你现有代码中每次循环就执行一次commit，会产生大量数据库IO请求，性能极低。建议循环内仅做数据写入/更新操作，循环结束后统一执行一次commit，异常时统一回滚即可。

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