python - 如何使用 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 发生。是否自动回滚交易？如果是这样，这可以停止吗？我不希望我的整个事务在出现问题时回滚，这就是为什么我首先 try catch 异常。

顺便说一句，我得到的错误消息是“sqlalchemy.exc.InternalError:(InternalError)当前事务被中止，命令被忽略直到事务 block 结束”，它发生在 update().execute()打电话。

Answer 1

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

一种解决方法是先执行更新。通过查看 cursor.rowcount 来检测它是否真的修改了一行；如果它没有修改任何行，则它不存在，INSERT 也不存在。 (当然，如果更新频率高于插入频率，这会更快。)

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

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

这对于生产质量代码有一个严重的问题:您必须准确地检测错误。据推测，您预计会遇到唯一约束检查​​，但您可能会遇到意想不到的事情，并且几乎不可能可靠地区分预期错误和意外错误。如果这不正确地遇到错误条件，则会导致模糊的问题，即不会更新或插入任何内容，也不会看到任何错误。对此要非常小心。您可以通过查看 Postgresql 的错误代码来缩小错误案例的范围，以确保它是您期望的错误类型，但潜在的问题仍然存在。

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