How can SQLAlchemy be taught to recover from a disconnect?

Question

According to http://docs.sqlalchemy.org/en/rel_0_9/core/pooling.html#disconnect-handling-pessimistic, SQLAlchemy can be instrumented to reconnect if an entry in the connection pool is no longer valid. I create the following test case to test this:

import subprocess
from sqlalchemy import create_engine, event
from sqlalchemy import exc
from sqlalchemy.pool import Pool

@event.listens_for(Pool, "checkout")
def ping_connection(dbapi_connection, connection_record, connection_proxy):
    cursor = dbapi_connection.cursor()
    try:
        print "pinging server"
        cursor.execute("SELECT 1")
    except:
        print "raising disconnect error"
        raise exc.DisconnectionError()
    cursor.close()

engine = create_engine('postgresql://postgres@localhost/test')

connection = engine.connect()

subprocess.check_call(['psql', str(engine.url), '-c',
    "select pg_terminate_backend(pid) from pg_stat_activity " +
    "where pid <> pg_backend_pid() " +
    "and datname='%s';" % engine.url.database],
    stdout=subprocess.PIPE)

result = connection.execute("select 'OK'")
for row in result:
    print "Success!", " ".join(row)

But instead of recovering I receive this exception:

sqlalchemy.exc.OperationalError: (OperationalError) terminating connection due to administrator command
server closed the connection unexpectedly
        This probably means the server terminated abnormally
        before or while processing the request.

Since "pinging server" is printed on the terminal it seems safe to conclude that the event listener is attached. How can SQLAlchemy be taught to recover from a disconnect?

Protractile

Answer 1

It looks like the checkout method is only called when you first get a connection from the pool (eg your connection = engine.connect() line)

If you subsequently lose your connection, you will have to explicitly replace it, so you could just grab a new one, and retry your sql:

try:
    result = connection.execute("select 'OK'")
except sqlalchemy.exc.OperationalError:  # may need more exceptions here
    connection = engine.connect()  # grab a new connection
    result = connection.execute("select 'OK'")  # and retry

This would be a pain to do around every bit of sql, so you could wrap database queries using something like:

def db_execute(conn, query):
    try:
        result = conn.execute(query)
    except sqlalchemy.exc.OperationalError:  # may need more exceptions here (or trap all)
        conn = engine.connect()  # replace your connection
        result = conn.execute(query)  # and retry
    return result

The following:

result = db_execute(connection, "select 'OK'")

Should now succeed.

Another option would be to also listen for the invalidate method, and take some action at that time to replace your connection.

Kr

You seem to have confirmed that the SQLAlchemy documentation is misleading. The Pool in this context of no value if every query needs to be wrapped in a helper method does does exception handling and explicit recovery. – Protractile

Answer 2

You can use the pre ping - this emits a test statement to make sure that the database connection is still viable before running the actual statement.

The approach adds a small bit of overhead to the connection checkout process, however is otherwise the most simple and reliable approach to completely eliminating database errors due to stale pooled connections.

engine = create_engine("mysql+pymysql://user:pw@host/db", pool_pre_ping=True)

sqlalchemy disconnect handling

Delay