有没有办法测试 SQLAlchemy 连接？

Question

我正在使用 SQLAlchemy 连接以将 pandas DataFrame 写入 MySQL 数据库。在我的代码的早期，我创建了一个 SQLAlchemy 引擎：

 engine = create_my_sqlalchemy_connection()

我执行一些查询，进行一些计算，稍后尝试使用相同的引擎写入数据库：

 df.to_sql('my_table', engine, if_exists='append', index=False)

有时这行得通，有时在代码准备好写入数据库时连接丢失，并且出现错误。

我可以尝试一下，但如果需要的话创建一个新连接：

 try:
    df.to_sql('my_table', engine, if_exists='append', index=False)
except:
    engine = create_my_sqlalchemy_connection()
    df.to_sql('my_table', engine, if_exists='append', index=False)

但是，我想我会伸出援手，看看是否有人知道更好的方法（例如，是否有一些我不知道的用于测试连接是否仍然存在的 SQLAlchemy 方法）。

原文由 mgig 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

您可以让 SQLAlchemy 使用参数 pool_pre_ping 检查连接的活跃度： https ://docs.sqlalchemy.org/en/13/core/engines.html#sqlalchemy.create_engine.params.pool_pre_ping

如果为 True 将启用连接池“预 ping”功能，该功能会在每次结帐时测试连接的活动性。

只需在创建引擎时使用即可启用它。

原文由 user787267 发布，翻译遵循 CC BY-SA 4.0 许可协议