假设我想将几条 SQL 记录上传到可能尚未填充的表中。如果在表中或要提交到表的记录中已经存在具有主键（“ID”）的记录，我想用新记录替换现有记录。 我用的是mssql，SQL server 2008。
我的第一个猜测是
try:
session.add(record)
session.commit
except:
session.query().\
filter(Class.ID == record.ID).\
update(some expression)
session.commit()
表达式应该是什么？有没有更清洁（更安全！）的方式来做到这一点？
【问题讨论】：
标签： python sql-server sql-server-2008 sqlalchemy
一般来说，除非使用保证原子性的语句，否则您总是必须考虑可能由多个参与者尝试插入或更新（不要忘记删除）引起的竞争条件。即使是MERGE statement，虽然是一条语句，但如果使用不当，也可以have race conditions。
传统上，这种“upsert”是使用存储过程或其他可用的 SQL 或实现特定功能（如 MERGE 语句）执行的。
如果出现完整性错误，SQLAlchemy 解决方案必须尝试插入并执行更新，或者如果没有行受到影响，则执行 udpate 并尝试插入。它应该准备好在两个操作都失败的情况下重试（一行可能会被删除或插入）：
from sqlalchemy.exc import IntegrityError
while True: # Infinite loop, use a retry counter if necessary
try:
# begin a save point, prevents the whole transaction failing
# in case of an integrity error
with session.begin_nested():
session.add(record)
# Flush instead of commit, we need the transaction intact
session.flush()
# If the flush is successful, break out of the loop as the insert
# was performed
break
except IntegrityError:
# Attempt the update. If the session has to reflect the changes
# performed by the update, change the `synchronize_session` argument.
if session.query(Class).\
filter_by(ID=record.ID).\
update({...},
syncronize_session=False):
# 1 or more rows were affected (hopefully 1)
break
# Nothing was updated, perhaps a DELETE in between
# Both operations have failed, retry
session.commit()
关于
如果在表中或要提交到表的记录中已经存在具有主键（“ID”）的记录，我想用新记录替换现有记录。
如果您可以确定不会对相关表进行并发更新，您可以使用Session.merge 来完成此类任务：
# Records have primary key set, on which merge can either load existing
# state and merge, or create a new record in session if none was found.
for record in records:
merged_record = session.merge(record)
# Note that merged_record is not record
session.commit()
SQLAlchemy 合并将首先检查具有给定主键的实例是否存在于身份映射中。如果没有，并且load 作为True 传递，它将检查数据库的主键。如果给定实例没有主键或找不到实例，则会创建一个新实例。
然后，合并会将给定实例的状态复制到定位/创建的实例上。返回新实例。
【讨论】：
Class.ID 是。该块是安全的，因为它使用了一个保存点，如果插入失败，它会回滚。错误被捕获并正确处理，因此也是安全的。任何其他错误都应导致回滚。
INSERT ... SELECT ... FROM ... WHERE NOT EXISTS () 查询）。这本质上是 2 个操作，并且会有竞争条件。你必须执行插入，看看它是否失败。
Session.merge（因为它检查本地身份映射+可选的数据库检查，您仍然有可能的比赛），但这取决于什么“一些表达”实际上，您的模型是如何设置的等等。
没有。这样做有一个更好的模式。首先进行查询以查看记录是否已存在，然后进行相应的操作。
使用您的语法，它将类似于以下内容：
result = session.query().filter(Class.ID == record.ID).first()
# If record does not exist in Db, then add record
if result is None:
try:
session.add(record)
session.commit()
except:
db.rollback()
log.error('Rolling back transaction in query-none block')
# If record does exist, then update value of record in Db
else:
try:
session.query().\
filter(Class.ID == record.ID).\
update(some expression)
session.commit()
except:
db.rollback()
log.error('Rolling back transaction')
将数据库操作包装在 try/except 块中通常是一个好主意，这样您就可以使用您所写内容的 try 部分。根据您正在执行的操作，except 块通常应该向您显示错误消息或执行数据库回滚。
【讨论】：