Add mechanisms to suspend connection replacement during transactions.

Author: Cito

DBUtils/PersistentDB.py

@@ -9,6 +9,10 @@
 Robustness is provided by using "hardened" SteadyDB connections.
 Even if the underlying database is restarted and all connections
 are lost, they will be automatically and transparently reopened.
+However, since you don't want this to happen in the middle of a database
+transaction, you must explicitly start transactions with the begin()
+method so that SteadyDB knows that the underlying connection shall not
+be replaced and errors passed on until the transaction is completed.
 
 Measures are taken to make the database connections thread-affine.
 This means the same thread always uses the same cached connection,
@@ -78,9 +82,14 @@
 the connection will be automatically closed when the thread dies.
 You can change this behavior be setting the closeable parameter.
 
-Note that by setting the threadlocal parameter to threading.local,
-getting connections may become a bit faster, but this may not work in
-all environments (for instance, mod_wsgi is known to cause problems
+Note that you need to explicitly start transactions by calling the
+begin() method. This ensures that the transparent reopening will be
+suspended until the end of the transaction, and that the connection
+will be rolled back before being reused by the same thread.
+
+By setting the threadlocal parameter to threading.local, getting
+connections may become a bit faster, but this may not work in all
+environments (for instance, mod_wsgi is known to cause problems
 since it clears the threading.local data between requests).
 
 

DBUtils/PersistentPg.py

@@ -9,6 +9,10 @@
 Robustness is provided by using "hardened" SteadyPg connections.
 Even if the underlying database is restarted and all connections
 are lost, they will be automatically and transparently reopened.
+However, since you don't want this to happen in the middle of a database
+transaction, you must explicitly start transactions with the begin()
+method so that SteadyPg knows that the underlying connection shall not
+be replaced and errors passed on until the transaction is completed.
 
 Measures are taken to make the database connections thread-affine.
 This means the same thread always uses the same cached connection,
@@ -68,9 +72,15 @@
 the connection will be automatically closed when the thread dies.
 You can change this behavior be setting the closeable parameter.
 
-Note that by setting the threadlocal parameter to threading.local,
-getting connections may become a bit faster, but this may not work in
-all environments (for instance, mod_wsgi is known to cause problems
+Note that you need to explicitly start transactions by calling the
+begin() method. This ensures that the transparent reopening will be
+suspended until the end of the transaction, and that the connection
+will be rolled back before being reused in the same thread. To end
+transactions, use on of the end(), commit() or rollback() methods.
+
+By setting the threadlocal parameter to threading.local, getting
+connections may become a bit faster, but this may not work in all
+environments (for instance, mod_wsgi is known to cause problems
 since it clears the threading.local data between requests).
 
 

DBUtils/PooledDB.py

@@ -10,6 +10,10 @@
 Robustness is provided by using "hardened" SteadyDB connections.
 Even if the underlying database is restarted and all connections
 are lost, they will be automatically and transparently reopened.
+However, since you don't want this to happen in the middle of a database
+transaction, you must explicitly start transactions with the begin()
+method so that SteadyDB knows that the underlying connection shall not
+be replaced and errors passed on until the transaction is completed.
 
 Measures are taken to make the pool of connections thread-safe.
 If the underlying DB-API module is thread-safe at the connection level,
@@ -48,6 +52,9 @@
         the connection is automatically reset (closed and reopened).
     setsession: an optional list of SQL commands that may serve to
         prepare the session, e.g. ["set datestyle to german", ...]
+    reset: how connections should be reset when returned to the pool
+        (False or None to rollback transcations started with begin(),
+        the default value True always issues a rollback for safety's sake)
     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
@@ -107,6 +114,12 @@
     cur.close() # or del cur
     db.close() # or del db
 
+Note that you need to explicitly start transactions by calling the
+begin() method. This ensures that the connection will not be shared
+with other threads, that the transparent reopening will be suspended
+until the end of the transaction, and that the connection will be rolled
+back before being given back to the connection pool.
+
 
 Ideas for improvement:
 
@@ -162,7 +175,7 @@ class PooledDB:
     def __init__(self, creator,
             mincached=0, maxcached=0,
             maxshared=0, maxconnections=0, blocking=False,
-            maxusage=None, setsession=None,
+            maxusage=None, setsession=None, reset=True,
             failures=None, ping=1,
             *args, **kwargs):
         """Set up the DB-API 2 connection pool.
@@ -188,6 +201,9 @@ def __init__(self, creator,
             the connection is automatically reset (closed and reopened).
         setsession: optional list of SQL commands that may serve to prepare
             the session, e.g. ["set datestyle to ...", "set time zone ..."]
+        reset: how connections should be reset when returned to the pool
+            (False or None to rollback transcations started with begin(),
+            True to always issue a rollback for safety's sake)
         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
@@ -215,6 +231,7 @@ def __init__(self, creator,
         self._args, self._kwargs = args, kwargs
         self._maxusage = maxusage
         self._setsession = setsession
+        self._reset = reset
         self._failures = failures
         self._ping = ping
         if mincached is None:
@@ -287,6 +304,12 @@ def connection(self, shareable=True):
                 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
+                    while con.con._transaction:
+                        # do not share connections which are in a transaction
+                        self._shared_cache.insert(0, con)
+                        self._condition.wait()
+                        self._shared_cache.sort()
+                        con = self._shared_cache.pop(0)
                     con.con._ping_check() # check the underlying connection
                     con.share() # increase share of this connection
                 # put the connection (back) into the shared cache
@@ -339,14 +362,8 @@ def cache(self, con):
         self._condition.acquire()
         try:
             if not self._maxcached or len(self._idle_cache) < self._maxcached:
-                # the idle cache is not full, so put it there, but
-                try: # before returning the connection back to the pool,
-                    con.rollback() # perform a rollback
-                    # in order to prevent uncommited actions from being
-                    # unintentionally commited by some other thread
-                except Exception:
-                    # if an error occurs (no transaction, not supported)
-                    pass # then it will be silently ignored
+                con._reset(force=self._reset) # rollback possible transaction
+                # the idle cache is not full, so put it there
                 self._idle_cache.append(con) # append it to the idle cache
             else: # if the idle cache is already full,
                 con.close() # then close the connection
@@ -440,9 +457,30 @@ def __init__(self, con):
         self.con = con
         self.shared = 1
 
-    def __cmp__(self, other):
-        """Compare how often the connections are shared."""
-        return self.shared - other.shared
+    def __lt__(self, other):
+        if self.con._transaction == other.con._transaction:
+            return self.shared < other.shared
+        else:
+            return not self.con._transaction
+
+    def __le__(self, other):
+        if self.con._transaction == other.con._transaction:
+            return self.shared == other.shared
+        else:
+            return not self.con._transaction
+
+    def __eq__(self, other):
+        return (self.con._transaction == other.con._transaction
+            and self.shared == other.shared)
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
+    def __gt__(self, other):
+        return other.__lt__(self)
+
+    def __ge__(self, other):
+        return other.__le__(self)
 
     def share(self):
         """Increase the share of this connection."""

DBUtils/PooledPg.py

@@ -10,6 +10,10 @@
 Robustness is provided by using "hardened" SteadyPg connections.
 Even if the underlying database is restarted and all connections
 are lost, they will be automatically and transparently reopened.
+However, since you don't want this to happen in the middle of a database
+transaction, you must explicitly start transactions with the begin()
+method so that SteadyPg knows that the underlying connection shall not
+be replaced and errors passed on until the transaction is completed.
 
 Measures are taken to make the pool of connections thread-safe
 regardless of the fact that the classic PyGreSQL pg module itself
@@ -79,6 +83,12 @@
     res = db.query(...).getresult()
     db.close() # or del db
 
+Note that you need to explicitly start transactions by calling the
+begin() method. This ensures that the transparent reopening will be
+suspended until the end of the transaction, and that the connection will
+be rolled back before being given back to the connection pool. To end
+transactions, use on of the end(), commit() or rollback() methods.
+
 
 Ideas for improvement:
 
@@ -131,7 +141,7 @@ class PooledPg:
     def __init__(self,
             mincached=0, maxcached=0,
             maxconnections=0, blocking=False,
-            maxusage=None, setsession=None,
+            maxusage=None, setsession=None, reset=None,
             *args, **kwargs):
         """Set up the PostgreSQL connection pool.
 
@@ -150,13 +160,17 @@ def __init__(self,
             the connection is automatically reset (closed and reopened).
         setsession: optional list of SQL commands that may serve to prepare
             the session, e.g. ["set datestyle to ...", "set time zone ..."]
+        reset: how connections should be reset when returned to the pool
+            (0 or None to rollback transcations started with begin(),
+            1 to always issue a rollback, 2 for a complete reset)
         args, kwargs: the parameters that shall be used to establish
             the PostgreSQL connections using class PyGreSQL pg.DB()
 
         """
         self._args, self._kwargs = args, kwargs
         self._maxusage = maxusage
         self._setsession = setsession
+        self._reset = reset or 0
         if mincached is None:
             mincached = 0
         if maxcached is None:
@@ -200,7 +214,15 @@ def connection(self):
     def cache(self, con):
         """Put a connection back into the pool cache."""
         try:
-            self._cache.put(con, 0)
+            if self._reset == 2:
+                con.reset() # reset the connection completely
+            else:
+                if self._reset or con._transaction:
+                    try:
+                        con.rollback() # rollback a possible transaction
+                    except Exception:
+                        pass
+            self._cache.put(con, 0) # and then put it back into the cache
         except Full:
             con.close()
         if self._connections:
@@ -255,7 +277,7 @@ def reopen(self):
         """Reopen the pooled connection."""
         # If the connection is already back in the pool,
         # get another connection from the pool,
-        # otherwise reopen the unerlying connection.
+        # otherwise reopen the underlying connection.
         if self._con:
             self._con.reopen()
         else: