Support ping() method for pooled and persistent connections.

Author: Cito

DBUtils/PersistentDB.py

@@ -39,6 +39,11 @@
     failures: an optional exception class or a tuple of exception classes
         for which the connection failover mechanism shall be applied,
         if the default (OperationalError, InternalError) is not adequate
+    ping: an optional flag controlling when connections are checked
+        with the ping() method if such a method is available
+        (0 = None = never, 1 = default = whenever it is requested,
+        2 = when a cursor is created, 4 = when a query is executed,
+        7 = always, and all other bit combinations of these values)
     closeable: if this is set to true, then closing connections will
         be allowed, but by default this will be silently ignored
     threadlocal: an optional class for representing thread-local data
@@ -129,7 +134,7 @@ class PersistentDB:
     version = __version__
 
     def __init__(self, creator,
-            maxusage=None, setsession=None, failures=None,
+            maxusage=None, setsession=None, failures=None, ping=1,
             closeable=False, threadlocal=None, *args, **kwargs):
         """Set up the persistent DB-API 2 connection generator.
 
@@ -143,6 +148,10 @@ def __init__(self, creator,
         failures: an optional exception class or a tuple of exception classes
             for which the connection failover mechanism shall be applied,
             if the default (OperationalError, InternalError) is not adequate
+        ping: determines when the connection should be checked with ping()
+            (0 = None = never, 1 = default = whenever it is requested,
+            2 = when a cursor is created, 4 = when a query is executed,
+            7 = always, and all other bit combinations of these values)
         closeable: if this is set to true, then closing connections will
             be allowed, but by default this will be silently ignored
         threadlocal: an optional class for representing thread-local data
@@ -168,6 +177,7 @@ def __init__(self, creator,
         self._maxusage = maxusage
         self._setsession = setsession
         self._failures = failures
+        self._ping = ping
         self._closeable = closeable
         self._args, self._kwargs = args, kwargs
         self.thread = (threadlocal or ThreadingLocal.local)()
@@ -176,7 +186,7 @@ def steady_connection(self):
         """Get a steady, non-persistent DB-API 2 connection."""
         return connect(self._creator,
             self._maxusage, self._setsession,
-            self._failures, self._closeable, None,
+            self._failures, self._ping, self._closeable,
             *self._args, **self._kwargs)
 
     def connection(self, shareable=False):
@@ -194,6 +204,7 @@ def connection(self, shareable=False):
             if not con.threadsafety():
                 raise NotSupportedError("Database module is not thread-safe.")
             self.thread.connection = con
+        con._ping_check()
         return con
 
     def dedicated_connection(self):

DBUtils/PooledDB.py

@@ -51,6 +51,11 @@
     failures: an optional exception class or a tuple of exception classes
         for which the connection failover mechanism shall be applied,
         if the default (OperationalError, InternalError) is not adequate
+    ping: an optional flag controlling when connections are checked
+        with the ping() method if such a method is available
+        (0 = None = never, 1 = default = whenever fetched from the pool,
+        2 = when a cursor is created, 4 = when a query is executed,
+        7 = always, and all other bit combinations of these values)
 
     The creator function or the connect function of the DB-API 2 compliant
     database module specified as the creator will receive any additional
@@ -157,7 +162,8 @@ class PooledDB:
     def __init__(self, creator,
             mincached=0, maxcached=0,
             maxshared=0, maxconnections=0, blocking=False,
-            maxusage=None, setsession=None, failures=None,
+            maxusage=None, setsession=None,
+            failures=None, ping=1,
             *args, **kwargs):
         """Set up the DB-API 2 connection pool.
 
@@ -185,6 +191,10 @@ def __init__(self, creator,
         failures: an optional exception class or a tuple of exception classes
             for which the connection failover mechanism shall be applied,
             if the default (OperationalError, InternalError) is not adequate
+        ping: determines when the connection should be checked with ping()
+            (0 = None = never, 1 = default = whenever fetched from the pool,
+            2 = when a cursor is created, 4 = when a query is executed,
+            7 = always, and all other bit combinations of these values)
         args, kwargs: the parameters that shall be passed to the creator
             function or the connection constructor of the DB-API 2 module
 
@@ -206,6 +216,7 @@ def __init__(self, creator,
         self._maxusage = maxusage
         self._setsession = setsession
         self._failures = failures
+        self._ping = ping
         if mincached is None:
             mincached = 0
         if maxcached is None:
@@ -247,7 +258,7 @@ def steady_connection(self):
         """Get a steady, unpooled DB-API 2 connection."""
         return connect(self._creator,
             self._maxusage, self._setsession,
-            self._failures, True, None,
+            self._failures, self._ping, True,
             *self._args, **self._kwargs)
 
     def connection(self, shareable=True):
@@ -269,11 +280,14 @@ def connection(self, shareable=True):
                         con = self._idle_cache.pop(0)
                     except IndexError: # else get a fresh connection
                         con = self.steady_connection()
+                    else:
+                        con._ping_check() # check this connection
                     con = SharedDBConnection(con)
                     self._connections += 1
                 else: # shared cache full or no more connections allowed
                     self._shared_cache.sort() # least shared connection first
                     con = self._shared_cache.pop(0) # get it
+                    con.con._ping_check() # check the underlying connection
                     con.share() # increase share of this connection
                 # put the connection (back) into the shared cache
                 self._shared_cache.append(con)
@@ -292,6 +306,8 @@ def connection(self, shareable=True):
                     con = self._idle_cache.pop(0)
                 except IndexError: # else get a fresh connection
                     con = self.steady_connection()
+                else:
+                    con._ping_check() # check connection
                 con = PooledDedicatedDBConnection(self, con)
                 self._connections += 1
             finally:

DBUtils/SteadyDB.py

@@ -103,8 +103,8 @@ class InvalidCursor(SteadyDBError):
     """Database cursor is invalid."""
 
 
-def connect(creator, maxusage=None, setsession=None, failures=None,
-        closeable=True, ping=None, *args, **kwargs):
+def connect(creator, maxusage=None, setsession=None,
+        failures=None, ping=1, closeable=True, *args, **kwargs):
     """A tough version of the connection constructor of a DB-API 2 module.
 
     creator: either an arbitrary function returning new DB-API 2 compliant
@@ -118,25 +118,27 @@ def connect(creator, maxusage=None, setsession=None, failures=None,
     failures: an optional exception class or a tuple of exception classes
         for which the failover mechanism shall be applied, if the default
         (OperationalError, InternalError) is not adequate
+    ping: determines when the connection should be checked with ping()
+        (0 = None = never, 1 = default = when _ping_check() is called,
+        2 = whenever a cursor is created, 4 = when a query is executed,
+        7 = always, and all other bit combinations of these values)
     closeable: if this is set to false, then closing the connection will
         be silently ignored, but by default the connection can be closed
-    ping: determines when the connection should be checked with ping()
-        (0 = default = never, 1 = on cursor(), 2 = on execute(), 3 = both)
     args, kwargs: the parameters that shall be passed to the creator
         function or the connection constructor of the DB-API 2 module
 
     """
-    return SteadyDBConnection(creator, maxusage, setsession, failures,
-        closeable, ping, *args, **kwargs)
+    return SteadyDBConnection(creator, maxusage, setsession,
+        failures, ping, closeable, *args, **kwargs)
 
 
 class SteadyDBConnection:
     """A "tough" version of DB-API 2 connections."""
 
     version = __version__
 
-    def __init__(self, creator, maxusage=None, setsession=None, failures=None,
-            closeable=True, ping=None, *args, **kwargs):
+    def __init__(self, creator, maxusage=None, setsession=None,
+            failures=None, ping=1, closeable=True, *args, **kwargs):
         """Create a "tough" DB-API 2 connection."""
         # basic initialization to make finalizer work
         self._con = None
@@ -176,8 +178,8 @@ def __init__(self, creator, maxusage=None, setsession=None, failures=None,
                 failures, tuple) and not issubclass(failures, Exception):
             raise TypeError("'failures' must be a tuple of exceptions.")
         self._failures = failures
+        self._ping = isinstance(ping, int) and ping or 0
         self._closeable = closeable
-        self._ping = ping or 0
         self._args, self._kwargs = args, kwargs
         self._store(self._create())
 
@@ -298,6 +300,38 @@ def _close(self):
                 pass
             self._closed = True
 
+    def _ping_check(self, ping=1, reconnect=True):
+        """Check whether the connection is still alive using ping().
+
+        If the the underlying connection is not active and the ping
+        parameter is set accordingly, the connection will be recreated.
+
+        """
+        if ping & self._ping:
+            try: # if possible, ping the connection
+                alive = self._con.ping()
+            except (AttributeError, IndexError, TypeError, ValueError):
+                self._ping = 0 # ping() is not available
+                alive = None
+                reconnect = False
+            except Exception:
+                alive = False
+            else:
+                if alive is None:
+                    alive = True
+                if alive:
+                    reconnect = False
+            if reconnect:
+                try: # try to reopen the connection
+                    con = self._create()
+                except Exception:
+                    pass
+                else:
+                    self._close()
+                    self._store(con)
+                    alive = True
+            return alive
+
     def dbapi(self):
         """Return the underlying DB-API 2 module of the connection."""
         if self._dbapi is None:
@@ -344,39 +378,29 @@ def _cursor(self, *args, **kwargs):
         """A "tough" version of the method cursor()."""
         # The args and kwargs are not part of the standard,
         # but some database modules seem to use these.
+        self._ping_check(2)
         try:
             if self._maxusage:
                 if self._usage >= self._maxusage:
                     # the connection was used too often
                     raise self._failure
-            con = self._con
-            if self._ping & 1:
-                try: # if possible, ping the connection
-                    ping = con.ping()
-                except (AttributeError, TypeError, ValueError):
-                    self._ping = 0 # ping() is not available
-                    ping = None
-                except Exception:
-                    ping = False
-                if ping is not None and not ping:
-                    raise self._failure # connection is dead
-            cursor = con.cursor(*args, **kwargs) # try to get a cursor
+            cursor = self._con.cursor(*args, **kwargs) # try to get a cursor
         except self._failures, error: # error in getting cursor
             try: # try to reopen the connection
-                con2 = self._create()
+                con = self._create()
             except Exception:
                 pass
             else:
                 try: # and try one more time to get a cursor
-                    cursor = con2.cursor(*args, **kwargs)
+                    cursor = con.cursor(*args, **kwargs)
                 except Exception:
                     pass
                 else:
                     self._close()
-                    self._store(con2)
+                    self._store(con)
                     return cursor
                 try:
-                    con2.close()
+                    con.close()
                 except Exception:
                     pass
             raise error # reraise the original error again
@@ -455,21 +479,12 @@ def _get_tough_method(self, name):
         def tough_method(*args, **kwargs):
             execute = name.startswith('execute')
             con = self._con
+            con._ping_check(4)
             try:
                 if con._maxusage:
                     if con._usage >= con._maxusage:
                         # the connection was used too often
                         raise con._failure
-                if con._ping & 2:
-                    try: # if possible, ping the connection
-                        ping = con._con.ping()
-                    except (AttributeError, TypeError, ValueError):
-                        self._ping = 0 # ping() is not available
-                        ping = None
-                    except Exception:
-                        ping = False
-                    if ping is not None and not ping:
-                        raise con._failure # connection is dead
                 if execute:
                     self._setsizes()
                 method = getattr(self._cursor, name)

DBUtils/Tests/TestPersistentDB.py

@@ -223,6 +223,58 @@ class threadlocal:
         persist = PersistentDB(dbapi, threadlocal=threadlocal)
         self.assert_(isinstance(persist.thread, threadlocal))
 
+    def test8_PingCheck(self):
+        Connection = dbapi.Connection
+        Connection.has_ping = True
+        Connection.num_pings = 0
+        persist = PersistentDB(dbapi, 0, None, None, 0, True)
+        db = persist.connection()
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 0)
+        db.close()
+        db = persist.connection()
+        self.assert_(not db._con.valid)
+        self.assertEqual(Connection.num_pings, 0)
+        persist = PersistentDB(dbapi, 0, None, None, 1, True)
+        db = persist.connection()
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 1)
+        db.close()
+        db = persist.connection()
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 2)
+        persist = PersistentDB(dbapi, 0, None, None, 2, True)
+        db = persist.connection()
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 2)
+        db.close()
+        db = persist.connection()
+        self.assert_(not db._con.valid)
+        self.assertEqual(Connection.num_pings, 2)
+        cursor = db.cursor()
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 3)
+        cursor.execute('select test')
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 3)
+        persist = PersistentDB(dbapi, 0, None, None, 4, True)
+        db = persist.connection()
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 3)
+        db.close()
+        db = persist.connection()
+        self.assert_(not db._con.valid)
+        self.assertEqual(Connection.num_pings, 3)
+        cursor = db.cursor()
+        db._con.close()
+        self.assert_(not db._con.valid)
+        self.assertEqual(Connection.num_pings, 3)
+        cursor.execute('select test')
+        self.assert_(db._con.valid)
+        self.assertEqual(Connection.num_pings, 4)
+        Connection.has_ping = False
+        Connection.num_pings = 0
+
 
 if __name__ == '__main__':
     unittest.main()