SQLAlchemy вече има страхотно описание как можете да използвате ping, за да бъдете песимистични относно свежестта на вашата връзка:
http://docs.sqlalchemy.org/en /latest/core/pooling.html#disconnect-handling-pessimistic
От там,
from sqlalchemy import exc
from sqlalchemy import event
from sqlalchemy.pool import Pool
@event.listens_for(Pool, "checkout")
def ping_connection(dbapi_connection, connection_record, connection_proxy):
cursor = dbapi_connection.cursor()
try:
cursor.execute("SELECT 1")
except:
# optional - dispose the whole pool
# instead of invalidating one at a time
# connection_proxy._pool.dispose()
# raise DisconnectionError - pool will try
# connecting again up to three times before raising.
raise exc.DisconnectionError()
cursor.close()
И тест, за да се уверите, че горното работи:
from sqlalchemy import create_engine
e = create_engine("mysql://scott:[email protected]/test", echo_pool=True)
c1 = e.connect()
c2 = e.connect()
c3 = e.connect()
c1.close()
c2.close()
c3.close()
# pool size is now three.
print "Restart the server"
raw_input()
for i in xrange(10):
c = e.connect()
print c.execute("select 1").fetchall()
c.close()
