Как эффективно выполнить массовую вставку или обновление с помощью SQLAlchemy?

Question

Я использую SQLAlchemy с бэкэндом Postgres, чтобы выполнить массовую вставку или обновление. Чтобы улучшить производительность, я пытаюсь зафиксировать только один раз каждые тысячи строк:

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 терпит неудачу, это оставляет вещи в странном состоянии, которое предотвращает ОБНОВЛЕНИЕ. Это автоматически откат транзакции? Если это так, можно ли это остановить? Я не хочу, чтобы вся транзакция откатывалась в случае возникновения проблемы, поэтому я пытаюсь поймать исключение в первую очередь.

К сожалению, я получаю сообщение об ошибке: «sqlalchemy.exc.InternalError: (InternalError) текущая транзакция отменена, команды игнорируются до конца блока транзакции», и это происходит при обновлении (). Execute () звоните.

Answer 1

Вы столкнулись с каким-то странным поведением, специфичным для Postgresql: если в транзакции происходит ошибка, она вызывает откат всей транзакции. Я считаю это ошибкой дизайна Postgres; в некоторых случаях для его обхода требуется немало SQL-искажения.

Один из обходных путей - сначала сделать ОБНОВЛЕНИЕ. Определите, действительно ли он изменил строку, посмотрев на cursor.rowcount; если он не изменял какие-либо строки, он не существовал, так что вставьте. (Это будет быстрее, если вы обновляете чаще, чем вставляете, конечно.)

Другой обходной путь - использование точек сохранения:

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

Это серьезная проблема для кода производственного качества: вы должны точно определить ошибку. Предположительно, вы ожидаете выполнить уникальную проверку ограничения, но вы можете выполнить что-то неожиданное, и может быть почти невозможно надежно отличить ожидаемую ошибку от неожиданной. Если это неправильно соответствует условию ошибки, это приведет к непонятным проблемам, при которых ничего не будет обновлено или вставлено, а ошибки не будет видно. Будьте очень осторожны с этим. Вы можете сузить число ошибок, посмотрев код ошибки Postgresql, чтобы убедиться, что вы ожидаете тип ошибки, но потенциальная проблема все еще существует.

Наконец, если вы действительно хотите выполнить пакетную вставку или обновление, вы на самом деле хотите выполнить многие из них за несколько команд, а не по одному элементу на команду. Для этого требуется более сложный SQL: SELECT, вложенный в INSERT, отфильтровывающий нужные элементы для вставки и обновления.

Answer 2

Эта ошибка из PostgreSQL. PostgreSQL не позволяет вам выполнять команды в одной и той же транзакции, если одна команда создает ошибку. Чтобы это исправить, вы можете использовать вложенные транзакции (реализованные с использованием точек сохранения SQL) через conn.begin_nested(). Вот что-то, что может сработать. Я сделал так, чтобы код использовал явные соединения, разложил часть на части и заставил код использовать менеджер контекста для правильного управления транзакциями.

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 может принимать список запросов, если все вставки используют одинаковые столбцы). Если вам нужно обрабатывать одновременные обновления, вам все равно придется выполнять обработку ошибок либо путем повторных попыток, либо отступая от одной к другой вставки.