Có cách nào để kiểm tra kết nối SQLAlchemy không?

Question

Tôi đang sử dụng SQLAlchemy để kết nối để viết một DataFrame gấu trúc vào cơ sở dữ liệu MySQL. Sớm trong mã của tôi tôi có thể tạo một động cơ SQLAlchemy:

engine = create_my_sqlalchemy_connection() 

tôi thực hiện một số truy vấn, làm một số tính toán, và sau đó cố gắng sử dụng động cơ tương tự ghi vào cơ sở dữ liệu một chút sau:

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

Đôi khi điều này làm việc, và đôi khi kết nối bị mất bởi thời gian mã đã sẵn sàng để ghi vào DB, và có một lỗi.

tôi có thể làm một thử, trừ và tạo một kết nối mới nếu cần thiết:

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) 

Tuy nhiên, tôi nghĩ rằng tôi muốn tiếp cận và xem có ai biết một cách tốt hơn (ví dụ như nếu có một số Phương pháp SQLAlchemy mà tôi không biết để kiểm tra xem kết nối vẫn tồn tại).

Answer 1

Rất đáng để thử thuộc tính 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)