有沒有辦法測試一個SQLAlchemy連接？

我正在使用SQLAlchemy連接將一個熊貓DataFrame寫入MySQL數據庫。早在我的代碼創建一個SQLAlchemy的發動機：有沒有辦法測試一個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方法，我不知道用於測試以查看連接是否仍然存在）。

來源

2015-10-04 mgig

你知道「稍後」是什麼嗎？秒，分鐘，小時？到目前爲止，我從來沒有遇到代碼運行幾分鐘的問題，但是如果df非常大，我想它可能會與更長的intervalls –

有所不同，您可以循環訪問df_chunks並在它們上調用to_sql（）？ – s5s

值得嘗試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)

來源

2017-10-16 16:42:05 Dan

有沒有辦法測試一個SQLAlchemy連接？

回答

相關問題