SQLAlchemy Connectionをテストする方法はありますか？

Question

SQLAlchemyを使用して、pandas DataFrameをMySQLデータベースに書き込むために接続しています。

engine = create_my_sqlalchemy_connection() 

は、私は、いくつかのクエリを実行し、いくつかの計算を行う、その後、少し後にデータベースに書き込むように、同じエンジンを使用しよう：

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

早い段階で私のコードで私はSQLAlchemyのエンジンを作成します

時にはこれが動作し、コードがDBに書き込む準備ができてエラーが発生するまでに接続が失われることがあります。私は、試してやる除いて、必要に応じて新しい接続を作成することができ

：

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

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)