Ошибка курсора sqlalchemy во время yield_per

Я получаю следующую ошибку:

Traceback (most recent call last):
    main()
    for item in session.query(Item).yield_per(10):
    fetch = cursor.fetchmany(self._yield_per)
    self.cursor, self.context)
    l = self.process_rows(self._fetchmany_impl(size))
    row = self._fetchone_impl()
    self.__buffer_rows()
    self.__rowbuffer = collections.deque(self.cursor.fetchmany(size))
sqlalchemy.exc.ProgrammingError: (ProgrammingError) named cursor isn't valid anymore None None

Я подозреваю, что вызов session.commit() вмешивается в .yield_per

sessionmaker_ = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = scoped_session(sessionmaker_)

def foo(item):
  # DO something to the item 
  session.add(item)
  session.commit()

def main():
  for item in session.query(Item).yield_per(5):
    foo(item)

Любая идея?

Ответ 1

Если вы не выбрали все строки из курсора DBAPI, тогда обычно бывает плохой вызов для вызова commit() для этого соединения с курсором. В этом случае psycopg2 (который я предполагаю, что DBAPI, на котором вы находитесь) не может поддерживать состояние именованного курсора (который он использует, когда вы хотите, чтобы строки, буферизованные сервером) по транзакции.

Одна вещь, которую вы обязательно должны изменить здесь, - это то, как часто вы совершаете. В идеале вы ничего не совершаете до завершения всей вашей операции. Сессия автоматически очистит данные по мере необходимости (ну, если вы включили автозапуск, который я бы рекомендовал), или вы можете вызвать flush(), чтобы заставить его, но это не зависит от фактического совершения транзакции. Все эти вызовы commit() сделают операцию намного менее эффективной, чем она должна быть, и, конечно, это мешает курсору для другого набора результатов. Если вы просто поместите один commit() в конец вашего цикла, вы сразу решите обе проблемы.

Если вам все еще нужно зафиксировать до завершения всей операции, или даже если нет, я бы предпочел работать в кусках, а не использовать yield_per(), что довольно хрупко. Рецепт на http://www.sqlalchemy.org/trac/wiki/UsageRecipes/WindowedRangeQuery показывает один из способов сделать это. DBAPI не очень подходят для работы с чрезвычайно большими результирующими наборами, хотя psycopg2 дает нам немного больше возможностей.

Ответ 2

Проблема выше может быть решена с помощью еще одного сеанса

sessionmaker_ = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = scoped_session(sessionmaker_)
cool_session = scoped_session(sessionmaker_)

def foo(item):
   # DO something to the item 
   session.add(item)
   session.commit()

def main():
    for item in cool_session.query(Item).yield_per(5):
    item = session.merge(item, load=False)
    foo(item)

Ответ 3

Другой вариант - использовать опцию курсора "WITH HOLD", когда она попадет в релиз sqlalchemy: https://bitbucket.org/zzzeek/sqlalchemy/issues/3667/support-postgresqls-with-hold-cursor

Обратите внимание на предостережение Скотта упоминает о том, что Postgres материализуют курсоры при первом фиксации.