python - 有没有办法测试 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 方法)。

Answer 1

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

if True will enable the connection pool “pre-ping” feature that tests connections for liveness upon each checkout.

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