有没有办法测试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

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

if True将启用连接池“预Ping”功能，该功能可在每次检出时测试连接的活跃性。

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

Answer 2

如果您在将Pandas Dataframe写入SQL服务器时遇到超时问题，那么您的Dataframe可能非常大，或者在您插入时数据库必须检查许多约束。

要解决此问题，您需要在Pandas命令中设置chunksize参数：

DataFrame.to_sql(name, con, schema=None, if_exists='fail', index=True, index_label=None, chunksize=None, dtype=None)

块大小：int，可选

每次将以此大小成批写入行。默认情况下，将一次写入所有行。

我不知道您有多少行，但是10000行可能是一个很好的值。这样做的问题是，如果现在写入失败，您将插入一些行，但不是全部=(并且您不知道是哪行。

Answer 3

值得尝试一下Connection.closed属性。

if engine.closed:
    engine = create_my_sqlalchemy_connection()
    df.to_sql('my_table', engine, if_exists='append', index=False)
else:
    df.to_sql('my_table', engine, if_exists='append', index=False)