SQLAlchemy 在使用 fetchall 方法迭代结果后提交 sql 执行

Posted 2023-03-23

【中文标题】SQLAlchemy 在使用 fetchall 方法迭代结果后提交 sql 执行

【英文标题】：SQLAlchemy commit sql execution after iterating through results with fetchall method

【发布时间】：2020-08-06 02:17:49

【问题描述】：

我在读取通过 SQLAlchemy 中的 OUTPUT 语句创建的结果时遇到问题。我最初在读取插入行的 primaryKey 时遇到问题，但能够使用以下代码将 autocommit 设置为 false 来使其工作：

sqlString = """ INSERT INTO workflow.table  (columns)
                OUTPUT Inserted.primaryKey
                VALUES (values)""".format(table=self.table, columns=columns, primaryKey=self.primaryKey, values=values)

    with self.dict_db['engine'].connect().execution_options(autocommit=False) as connection:
      result = connection.execute(sqlString)
      primaryKey = result.fetchone()[0]
      result.cursor.commit()

现在我正在编写一个删除语句，我想做类似的事情，但是在遍历我的结果之后，游标对象设置为无，所以我不能再提交。我已经尝试过使用迭代 results 和 results.fetchall() 的 for 循环，在这两种情况下我都无法提交，因为光标是 None。这是我的代码：

    sqlString = """ DELETE FROM workflow.table OUTPUT Deleted.primaryKey where """.format(table=self.table, primaryKey=self.primaryKey, where=whereStatement)

    with self.dict_db['engine'].connect().execution_options(autocommit=False) as connection:
      result = connection.execute(sqlString)
      # cursor exists
      primaryKeyList = [item[0] for item in result.fetchall()]
      # cursor is now None
      result.cursor.commit()

这行不通的事实让我重新思考两个数据库执行。我在这里做错了什么还是我缺少一些小语法？

注意：self.dict_db['engine'] 是使用 sqlalchemy create_engine 创建的

【参考方案1】：

使用连接创建事务（使用connection.begin() 并提交。

with self.dict_db['engine'].connect().execution_options(autocommit=False) as connection:
  txn = connection.begin()
  result = txn.execute(sqlString)
  # I'm not sure a cursor exists here, the result doesn't need one for fetchall()
  primaryKeyList = [item[0] for item in result.fetchall()]
  txn.commit()

更好的是，从connection.begin() 返回的对象实现了上下文管理器协议，因此您可以使用这个更简单的版本，并确保在没有异常的情况下提交事务，或者在有异常的情况下回滚：

with self.dict_db['engine'].connect().begin() as connection: 
  # connection here is really a Transaction, but all the connection methods work the same
  result = connection.execute(sqlString)
  primaryKeyList = [item[0] for item in result.fetchall()]
  # transaction is committed as the with block exits

有关更多详细信息，请参阅 SQLAlchemy 文档中的 Using Transactions。