SQLAlchemy Bağlantısını test etmenin bir yolu var mı?

Question

Bir MySQL veritabanına bir DataFrame pandas yazmak için bağlanmak için SQLAlchemy kullanıyorum.

engine = create_my_sqlalchemy_connection() 

Ben bazı sorguları yürütmek bazı hesaplamalar yapmak ve sonra bir süre sonra veritabanına yazmak için aynı motoru kullanmayı deneyin:

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

Erken benim kodda ben bir SQLAlchemy motoru oluşturmak

Bazen bu çalışır ve bazen kod DB'ye yazmaya hazır olduğunda zaman kaybedilir ve bir hata vardır.

ben hariç bir deneyin yapmak ve gerekirse yeni bir bağlantı oluşturabilir: Ben uzanmak ve herkes daha iyi bir yol bilen olmadığını görmek düşündüm,

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) 

Ancak (orada örneğin, bazı Bağlantının devam edip etmediğini görmek için sınamanın farkında olmadığım SQLAlchemy yöntemi.

Answer 1

Connection.closed özelliğini denemeye değer.

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)