SQLAlchemy के पास अब इस बारे में एक अच्छा लेखन है कि आप अपने कनेक्शन की ताजगी के बारे में निराशावादी होने के लिए पिंगिंग का उपयोग कैसे कर सकते हैं:
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()