Integrate "pre-ping" into connection pool.

Added native "pessimistic disconnection" handling to the :class:`.Pool`
object.  The new parameter :paramref:`.Pool.pre_ping`, available from
the engine as :paramref:`.create_engine.pool_pre_ping`, applies an
efficient form of the "pre-ping" recipe featured in the pooling
documentation, which upon each connection check out, emits a simple
statement, typically "SELECT 1", to test the connection for liveness.
If the existing connection is no longer able to respond to commands,
the connection is transparently recycled, and all other connections
made prior to the current timestamp are invalidated.

Change-Id: I89700d0075e60abd2250e54b9bd14daf03c71c00
Fixes: #3919

3 files changed, 32 insertions, 1 deletions

diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index bd8b7e68a..32e84c53f 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -355,6 +355,16 @@ def create_engine(*args, **kwargs):
        "sqlalchemy.pool" logger. Defaults to a hexstring of the object's
        id.
 
+    :param pool_pre_ping: boolean, if True will enable the connection pool
+        "pre-ping" feature that tests connections for liveness upon
+        each checkout.
+
+        .. versionadded:: 1.2
+
+        .. seealso::
+
+            :ref:`pool_disconnects_pessimistic`
+
     :param pool_size=5: the number of connections to keep open
         inside the connection pool. This used with
         :class:`~sqlalchemy.pool.QueuePool` as
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index dc2bd5f97..c7d574a21 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -459,6 +459,26 @@ class DefaultDialect(interfaces.Dialect):
     def do_close(self, dbapi_connection):
         dbapi_connection.close()
 
+    @util.memoized_property
+    def _dialect_specific_select_one(self):
+        return str(expression.select([1]).compile(dialect=self))
+
+    def do_ping(self, dbapi_connection):
+        cursor = None
+        try:
+            cursor = dbapi_connection.cursor()
+            try:
+                cursor.execute(self._dialect_specific_select_one)
+            finally:
+                cursor.close()
+        except self.dbapi.Error as err:
+            if self.is_disconnect(err, dbapi_connection, cursor):
+                return False
+            else:
+                raise
+        else:
+            return True
+
     def create_xid(self):
         """Create a random two-phase transaction ID.
 
diff --git a/lib/sqlalchemy/engine/strategies.py b/lib/sqlalchemy/engine/strategies.py
index 81bb2c508..286757698 100644
--- a/lib/sqlalchemy/engine/strategies.py
+++ b/lib/sqlalchemy/engine/strategies.py
@@ -121,7 +121,8 @@ class DefaultEngineStrategy(EngineStrategy):
                          'recycle': 'pool_recycle',
                          'events': 'pool_events',
                          'use_threadlocal': 'pool_threadlocal',
-                         'reset_on_return': 'pool_reset_on_return'}
+                         'reset_on_return': 'pool_reset_on_return',
+                         'pre_ping': 'pool_pre_ping'}
             for k in util.get_cls_kwargs(poolclass):
                 tk = translate.get(k, k)
                 if tk in kwargs: