如何向MySQL数据库新增数据并更新time字段重复的行？

针对你定期同步外部数据到本地MySQL，当time字段重复时更新该行、否则插入的需求，以下是两种实用实现方案，优先推荐批量处理方案以提升效率：

1. 批量UPSERT（推荐，适合大量数据） #

这种方式利用MySQL原生的INSERT ... ON DUPLICATE KEY UPDATE语法，通过SQLAlchemy的MySQL方言支持实现批量操作，效率远高于逐条处理。

步骤1：定义数据模型 #

首先必须确保time字段是主键或带有唯一约束，这样数据库才能识别重复行：

from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import UniqueConstraint

Base = declarative_base()

class Data(Base):
    __tablename__ = 'data_table'
    
    # 方案一：将time设为主键
    time = Column(Integer, primary_key=True)
    a = Column(Integer)
    b = Column(Integer)
    c = Column(Integer)
    
    # 方案二：如果time不能作为主键，添加唯一约束
    # __table_args__ = (UniqueConstraint('time', name='uq_data_time'),)

步骤2：批量执行UPSERT #

假设你从外部获取的新数据是字典列表，直接构造批量插入更新语句：

from sqlalchemy import create_engine
from sqlalchemy.dialects.mysql import insert
from sqlalchemy.orm import sessionmaker

# 初始化数据库连接
engine = create_engine('mysql+pymysql://your_username:your_password@localhost/your_db')
Session = sessionmaker(bind=engine)
session = Session()

# 模拟从外部获取的新数据
new_data_list = [
    {'time': 2, 'a': 4, 'b': 4, 'c': 5},
    {'time': 3, 'a': 2, 'b': 3, 'c': 4},
    {'time': 4, 'a': 4, 'b': 4, 'c': 4}
]

# 构造插入语句，指定重复时更新的字段
insert_stmt = insert(Data).values(new_data_list)
update_stmt = insert_stmt.on_duplicate_key_update(
    a=insert_stmt.inserted.a,
    b=insert_stmt.inserted.b,
    c=insert_stmt.inserted.c
)

# 执行并提交
session.execute(update_stmt)
session.commit()
session.close()

2. 单条数据UPSERT（适合少量数据） #

如果每次同步的数据量很小，可以使用SQLAlchemy的session.merge()方法，它会自动判断数据是否存在：

# 初始化会话（同上）
session = Session()

# 单条数据示例
single_data = Data(time=2, a=4, b=4, c=5)
# merge会自动查询：如果time存在则更新，不存在则插入
merged_data = session.merge(single_data)

session.commit()
session.close()

关键注意事项 #

• 唯一约束必须存在：无论是主键还是单独的唯一约束，time字段必须具备唯一性，否则ON DUPLICATE KEY UPDATE或merge无法识别重复行。
• 驱动适配：根据你使用的MySQL驱动调整连接字符串，比如mysql+mysqldb://（mysqlclient）或mysql+pymysql://（pymysql）。
• 字段匹配：确保新数据字典的键与模型字段名完全一致，避免映射错误。

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