如何使用SQLAlchemy有效地执行批量插入或更新？

Question

我正在使用带有Postgres后端的SQLAlchemy进行批量插入或更新.为了提高性能,我试图每千行左右只提交一次:

trans = engine.begin()
  for i, rec in enumerate(records):
    if i % 1000 == 0:
      trans.commit()
      trans = engine.begin()
    try:
        inserter.execute(...)
    except sa.exceptions.SQLError:
        my_table.update(...).execute()
trans.commit()

但是,这不起作用.似乎当INSERT失败时,它会使事物处于一种奇怪的状态,阻止UPDATE发生.它会自动回滚交易吗？如果是这样,可以停止吗？我不希望在发生问题时回滚整个事务,这就是为什么我首先尝试捕获异常.

我得到的错误消息BTW是"sqlalchemy.exc.InternalError:(InternalError)当前事务被中止,命令被忽略直到事务块结束",并且它发生在update().execute()调用上.

Answer 1

您正在遇到一些奇怪的Postgresql特定行为:如果事务中发生错误,它会强制回滚整个事务.我认为这是一个Postgres设计错误; 在某些情况下,需要相当多的SQL扭曲才能解决.

一种解决方法是首先执行UPDATE.通过查看cursor.rowcount来检测它是否实际修改了一行; 如果它没有修改任何行,它就不存在,所以INSERT也是如此.(当然,如果您更新频率比插入频率更快,这将会更快.)

另一种解决方法是使用保存点:

SAVEPOINT a;
INSERT INTO ....;
-- on error:
ROLLBACK TO SAVEPOINT a;
UPDATE ...;
-- on success:
RELEASE SAVEPOINT a;

这对于生产质量代码存在严重问题:您必须准确检测错误.据推测,您可能会遇到一个独特的约束检查,但是您可能会遇到意外情况,并且可能几乎无法可靠地将预期错误与意外错误区分开来.如果这不正确地击中错误条件,它将导致模糊的问题,没有任何内容将被更新或插入,并且不会看到错误.要非常小心.您可以通过查看Postgresql的错误代码来缩小错误情况,以确保它是您期望的错误类型,但潜在的问题仍然存在.

最后,如果你真的想要批量插入或更新,你实际上想要在几个命令中做很多,而不是每个命令一个项目.这需要更棘手的SQL:SELECT嵌套在INSERT中,过滤掉要插入和更新的正确项目.