Czy istnieje sposób na przetestowanie połączenia SQLAlchemy?

Question

Używam SQLAlchemy do połączenia, aby wypisać pandas DataFrame do bazy danych MySQL. Wcześnie w moim kodu utworzyć silnik sqlalchemy:

engine = create_my_sqlalchemy_connection() 

mogę wykonać kilka zapytań, zrobić kilka obliczeń, a następnie spróbuj użyć tego samego silnika do zapisu w bazie danych trochę później:

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

Czasami to działa, a czasami połączenie zostaje utracone do czasu, gdy kod jest gotowy do zapisu do bazy danych i wystąpił błąd.

mogłem zrobić spróbować, z wyjątkiem i utworzyć nowe połączenie w razie potrzeby:

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) 

Jednak myślałem, że dotrzeć i sprawdzić, czy ktoś zna lepszy sposób (na przykład, jeśli istnieją pewne Metoda SQLAlchemy, której nie znam do testowania, aby sprawdzić, czy połączenie nadal istnieje).

Answer 1

Warto wypróbować właściwość 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)