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?

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.