在MySQL中使用SQLAlchemy无法处理on_conflict_do_nothing问题

当前代码使用ORM的db.add_all()方法批量插入数据，遇到唯一键冲突时会直接抛出异常，无法跳过冲突行。要实现跳过冲突数据，需要利用MySQL的冲突处理语法，结合SQLAlchemy核心层的insert语句来实现，以下是两种可行方案：

方案一：使用INSERT IGNORE跳过冲突行 #

这种方式会忽略触发唯一键约束的行，继续插入其余数据。

修改后的代码：

from sqlalchemy.orm import Session
from sqlalchemy import insert

from app.config import config
from app.datastores.mysql.db import get_db
from app.datastores.mysql.models import models, schemas

async def adding_cronjob_alerts(
    alert: schemas.CreateAlert, generate_uuid, db: Session = Depends(get_db)
):
    logger.debug(f"[{generate_uuid}] - Customer Alert Doesn't Exist In DB => Storing")
    # 构造批量插入的字典数据（无需创建ORM模型实例）
    alerts_data = [item.dict() for item in alert]
    logger.debug(f"[{generate_uuid}] - Adding Customer Alerts Record To DB")
    # 构造带IGNORE前缀的insert语句
    insert_stmt = insert(models.ReaqtaAlerts).prefix_with('IGNORE').values(alerts_data)
    # 执行插入
    db.execute(insert_stmt)
    logger.debug(f"[{generate_uuid}] - Alerts Added Successfully")
    logger.debug(f"[{generate_uuid}] - Committing DB Session To MySQL Database")
    db.commit()
    logger.debug(f"[{generate_uuid}] - {CLOSING_DB_SESSION_STRING}")
    db.close()

方案二：使用ON DUPLICATE KEY UPDATE（可选更新或无操作） #

如果需要在冲突时更新字段，或者仅需跳过冲突（通过设置无意义的更新操作），可以使用该方式：

# 构造insert语句，冲突时可选择更新字段或无操作
insert_stmt = insert(models.ReaqtaAlerts).values(alerts_data).on_duplicate_key_update(
    # 示例1：更新updated_at字段为当前时间
    # updated_at=func.now(),
    # 示例2：无操作（仅跳过冲突）
    id=models.ReaqtaAlerts.id
)
db.execute(insert_stmt)

注意事项 #

• 确保ReaqtaAlerts模型定义了唯一键约束（主键、unique=True的字段），否则MySQL无法检测到冲突。
• 使用SQLAlchemy核心层的insert语句比ORM的add_all()性能更高，避免了创建大量模型实例的开销。
• 该方案适用于SQLAlchemy 1.4及以上版本，以及MySQL 5.1+版本。

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