Is there a way to test a SQLAlchemy connection?

Question

I am using SQLAlchemy to connect to write pandas DataFrame to MySQL database. At the beginning of my code, I create the SQLAlchemy mechanism:

engine = create_my_sqlalchemy_connection() 

I do some queries, do some calculations, and then try to use the same engine to write to the database a bit later:

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

Sometimes this works, and sometimes the connection is lost by the time the code is ready to be written to the database, and there is an error.

I could try, except and if necessary create a new connection:

 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) 

However, I thought I would get to you and see if anyone knows a better way (for example, if there is some SQLAlchemy method that I don’t know about for testing to find out if the connection still exists).

Answer 1

It is worth trying the Connection.closed property.

 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) 

Answer 2

If you have timeout problems when writing Pandas Dataframe data to an SQL server, your Dataframe is probably large enough or there are many limitations that the database should check when pasting.
To get around this, you need to set the chunksize argument in the Pandas command:
DataFrame.to_sql(name, con, schema=None, if_exists='fail', index=True, index_label=None, chunksize=None, dtype=None)

chunksize : int, optional
Rows will be recorded in parts of this size at a time. By default, all lines will be recorded at the same time.

I don't know how many rows you have, but 10000 is probably a good value. The problem is that if your record failed, you will insert a few lines, but not all = (and you won’t know which ones.

Answer 3

This may be useful for you as it seems your connection has expired. This is for those who, like me, have found this useful.

From the SQLAlchemy documentation when working with MySQL:

MySQL has a feature to automatically close the connection for connections that have been idle for a fixed period of time, the default is eight hours. To work around this problem, use the create_engine.pool_recycle parameter, which ensures that the connection is dropped and replaced with a new one if it was present in the pool for a fixed number of seconds:

engine = create_engine('mysql+mysqldb://...', pool_recycle=3600)