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

【问题标题】：How do I efficiently do a bulk insert-or-update with SQLAlchemy?如何使用 SQLAlchemy 有效地进行批量插入或更新？
【发布时间】：2009-08-25 19:38:02
【问题描述】：

我正在使用带有 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 的发生。它会自动回滚事务吗？如果是这样，这可以停止吗？我不希望我的整个事务在出现问题时回滚，这就是我首先尝试捕获异常的原因。

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

【问题讨论】：

标签： python sqlalchemy

【解决方案1】：

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

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

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

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

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

最后，如果你真的想做批量插入或更新，你实际上想在几个命令中完成其中的许多，而不是每个命令一个项目。这需要更复杂的 SQL：SELECT 嵌套在 INSERT 中，过滤掉要插入和更新的正确项目。

【讨论】：

“如果事务中发生错误，它会强制回滚整个事务。我认为这是 Postgres 设计错误。” - 这不是交易的重点吗？来自Wikipedia：“事务提供了一个‘全有或全无’的命题，说明在数据库中执行的每个工作单元必须要么完整完成，要么没有任何效果。”
@Spiffytech 反应良好。这实际上让我哈哈大笑。

【解决方案2】：

此错误来自 PostgreSQL。如果一个命令产生错误，PostgreSQL 不允许您在同一事务中执行命令。要解决此问题，您可以通过conn.begin_nested() 使用嵌套事务（使用 SQL 保存点实现）。这是可能有用的东西。我让代码使用显式连接，分解出分块部分，让代码使用上下文管理器来正确管理事务。

from itertools import chain, islice
def chunked(seq, chunksize):
    """Yields items from an iterator in chunks."""
    it = iter(seq)
    while True:
        yield chain([it.next()], islice(it, chunksize-1))

conn = engine.commit()
for chunk in chunked(records, 1000):
    with conn.begin():
        for rec in chunk:
            try:
                with conn.begin_nested():
                     conn.execute(inserter, ...)
            except sa.exceptions.SQLError:
                conn.execute(my_table.update(...))

尽管由于嵌套事务开销，这仍然不会有出色的性能。如果您想要更好的性能，请尝试使用 select 查询预先检测哪些行会产生错误并使用 executemany 支持（如果所有插入使用相同的列，execute 可以获取字典列表）。如果您需要处理并发更新，您仍然需要通过重试或回退到一一插入来进行错误处理。

【讨论】：