处理 sqlalchemy 断开连接的更好方法答案

【问题标题】：Better approach to handling sqlalchemy disconnects处理 sqlalchemy 断开连接的更好方法
【发布时间】：2015-06-20 11:38:17
【问题描述】：

我们一直在试验 sqlalchemy 的断开连接处理，以及它如何与 ORM 集成。我们研究了文档，建议似乎是捕获断开连接异常，发出rollback() 并重试代码。

例如：

import sqlalchemy as SA

retry = 2
while retry:
    retry -= 1
    try:
        for name in session.query(Names):
            print name
        break
    except SA.exc.DBAPIError as exc:
        if retry and exc.connection_invalidated:
            session.rollback()
        else:
            raise

我遵循基本原理 - 您必须回滚任何活动事务并重放它们以确保您的操作顺序一致。

但是——这意味着要在每个想要处理数据的函数中添加大量额外代码。此外，对于SELECT，我们并没有修改数据，回滚/重新请求的概念不仅难看，而且违反了DRY原则（不要重复）。

我想知道其他人是否介意分享他们如何使用 sqlalchemy 处理断开连接。

仅供参考：我们使用的是 sqlalchemy 0.9.8 和 Postgres 9.2.9

【问题讨论】：

目前我们正在使用Pessimistic Disconnect Handling 并取得一些成功来缓解MySQL has gone away。我们仍然在生产中看到一个案例，尽管我们似乎无法从这种情况中恢复，并且事务无法回滚并被卡住。虽然这可能与我们正在加入两个事务（ZODB 和 SQL）并且尚未使用 Two-Phase commits 的事实有关。
使用 PostgreSQL，我们还没有出现任何断开连接的情况，因此没有经验。
那么——你决定接受 try/catch/retry 逻辑了吗？我们的 ORM 类中有几十个查询函数，我们管理着几十个类。这真的加起来了。顺便说一句——直到最近，当 RHEL oom-killer 杀死了一个长时间运行的 postmaster 时，我们在重启 Postgres 时没有遇到任何问题。突然意识到我们需要优雅地从中恢复过来。
否，使用pessimistic disconnect handling 注册一个 SQLAlchemy 事件处理程序，每次从池中检出连接时都会运行该事件处理程序，并在返回之前通过对其执行SELECT 1 来验证连接要使用的。但是，正如文档所述，“以从池中签出的每个连接发出一些额外的 SQL 为代价”.
我没有遵循关于什么构成 checkout 事件的解释。假设您打开了一个连接，并且您正在查询一些数据。在处理过程中，postmaster 进程被终止。 SELECT 1 有什么帮助？还是他们说它在每个查询之前执行？

标签： python postgresql sqlalchemy

【解决方案1】：

我喜欢解决这个问题的方法是将我所有的数据库代码放在一个 lambda 或闭包中，并将其传递给一个辅助函数，该函数将处理捕获断开连接异常并重试。

你的例子：

import sqlalchemy as SA

def main():
    def query():
        for name in session.query(Names):
            print name

    run_query(query)

def run_query(f, attempts=2):
    while attempts > 0:
        attempts -= 1
        try:
            return f() # "break" if query was successful and return any results
         except SA.exc.DBAPIError as exc:
            if attempts > 0 and exc.connection_invalidated:
                session.rollback()
            else:
                raise

您可以通过将布尔值传递给 run_query 来处理您只进行读取并因此想要重试而不回滚的情况，从而使这更有趣。

这有助于您满足 DRY 原则，因为用于管理重试和回滚的所有难看的样板代码都放在一个位置。

【讨论】：

这是一个很好的解决方案，但现在我正在研究 SQL Alchemy 文档，看看它是否可以自动应用于每个查询，而不必将其传递给 run_query。
我还没有找到比猴子修补查询对象上的所有函数以自动应用它更好的解决方案。我仍然希望通过 SQL Alchemy 事件侦听器找到解决方案。看起来新的 pool_pre_ping=True 可能会在较新版本的 SQL alchemy 中解决它。
包装器比辅助函数更好。

【解决方案2】：

使用指数退避 (https://github.com/litl/backoff)：

@backoff.on_exception(
    backoff.expo,
    sqlalchemy.exc.DBAPIError,
    factor=7,
    max_tries=3,
    on_backoff=lambda details: LocalSession.get_main_sql_session().rollback(),
    on_giveup=lambda details: LocalSession.get_main_sql_session().flush(),  # flush the session
    logger=logging
)
def pessimistic_insertion(document_metadata):

    LocalSession.get_main_sql_session().add(document_metadata)
    LocalSession.get_main_sql_session().commit()

假设LocalSession.get_main_sql_session() 返回一个单例。

【讨论】：