- New features added to support engine/pool plugins with advanced

functionality.   Added a new "soft invalidate" feature to the
connection pool at the level of the checked out connection wrapper
as well as the :class:`._ConnectionRecord`.  This works similarly
to a modern pool invalidation in that connections aren't actively
closed, but are recycled only on next checkout; this is essentially
a per-connection version of that feature.  A new event
:class:`.PoolEvents.soft_invalidate` is added to complement it.
fixes #3379

- Added new flag
:attr:`.ExceptionContext.invalidate_pool_on_disconnect`.
Allows an error handler within :meth:`.ConnectionEvents.handle_error`
to maintain a "disconnect" condition, but to handle calling invalidate
on individual connections in a specific manner within the event.

- Added new event :class:`.DialectEvents.do_connect`, which allows
interception / replacement of when the :meth:`.Dialect.connect`
hook is called to create a DBAPI connection.  Also added
dialect plugin hooks :meth:`.Dialect.get_dialect_cls` and
:meth:`.Dialect.engine_created` which allow external plugins to
add events to existing dialects using entry points.
fixes #3355

1 files changed, 73 insertions, 13 deletions

diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 999cc1120..902309d75 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -219,6 +219,7 @@ class Pool(log.Identified):
         log.instance_logger(self, echoflag=echo)
         self._threadconns = threading.local()
         self._creator = creator
+        self._wrapped_creator = self._maybe_wrap_callable(creator)
         self._recycle = recycle
         self._invalidate_time = 0
         self._use_threadlocal = use_threadlocal
@@ -249,6 +250,32 @@ class Pool(log.Identified):
             for l in listeners:
                 self.add_listener(l)
 
+    def _maybe_wrap_callable(self, fn):
+        """Detect if creator accepts a single argument, or is sent
+        as a legacy style no-arg function.
+
+        """
+
+        try:
+            argspec = util.get_callable_argspec(fn, no_self=True)
+        except TypeError:
+            return lambda ctx: fn()
+
+        defaulted = argspec[3] is not None and len(argspec[3]) or 0
+        positionals = len(argspec[0]) - defaulted
+
+        # look for the exact arg signature that DefaultStrategy
+        # sends us
+        if (argspec[0], argspec[3]) == (['connection_record'], (None,)):
+            return fn
+        # or just a single positional
+        elif positionals == 1:
+            return fn
+        # all other cases, just wrap and assume legacy "creator" callable
+        # thing
+        else:
+            return lambda ctx: fn()
+
     def _close_connection(self, connection):
         self.logger.debug("Closing connection %r", connection)
         try:
@@ -428,6 +455,8 @@ class _ConnectionRecord(object):
 
     """
 
+    _soft_invalidate_time = 0
+
     @util.memoized_property
     def info(self):
         """The ``.info`` dictionary associated with the DBAPI connection.
@@ -476,7 +505,7 @@ class _ConnectionRecord(object):
         if self.connection is not None:
             self.__close()
 
-    def invalidate(self, e=None):
+    def invalidate(self, e=None, soft=False):
         """Invalidate the DBAPI connection held by this :class:`._ConnectionRecord`.
 
         This method is called for all connection invalidations, including
@@ -484,6 +513,13 @@ class _ConnectionRecord(object):
         :meth:`.Connection.invalidate` methods are called, as well as when any
         so-called "automatic invalidation" condition occurs.
 
+        :param e: an exception object indicating a reason for the invalidation.
+
+        :param soft: if True, the connection isn't closed; instead, this
+         connection will be recycled on next checkout.
+
+         .. versionadded:: 1.0.3
+
         .. seealso::
 
             :ref:`pool_connection_invalidation`
@@ -492,22 +528,31 @@ class _ConnectionRecord(object):
         # already invalidated
         if self.connection is None:
             return
-        self.__pool.dispatch.invalidate(self.connection, self, e)
+        if soft:
+            self.__pool.dispatch.soft_invalidate(self.connection, self, e)
+        else:
+            self.__pool.dispatch.invalidate(self.connection, self, e)
         if e is not None:
             self.__pool.logger.info(
-                "Invalidate connection %r (reason: %s:%s)",
+                "%sInvalidate connection %r (reason: %s:%s)",
+                "Soft " if soft else "",
                 self.connection, e.__class__.__name__, e)
         else:
             self.__pool.logger.info(
-                "Invalidate connection %r", self.connection)
-        self.__close()
-        self.connection = None
+                "%sInvalidate connection %r",
+                "Soft " if soft else "",
+                self.connection)
+        if soft:
+            self._soft_invalidate_time = time.time()
+        else:
+            self.__close()
+            self.connection = None
 
     def get_connection(self):
         recycle = False
         if self.connection is None:
-            self.connection = self.__connect()
             self.info.clear()
+            self.connection = self.__connect()
             if self.__pool.dispatch.connect:
                 self.__pool.dispatch.connect(self.connection, self)
         elif self.__pool._recycle > -1 and \
@@ -523,11 +568,18 @@ class _ConnectionRecord(object):
                 self.connection
             )
             recycle = True
+        elif self._soft_invalidate_time > self.starttime:
+            self.__pool.logger.info(
+                "Connection %r invalidated due to local soft invalidation; " +
+                "recycling",
+                self.connection
+            )
+            recycle = True
 
         if recycle:
             self.__close()
-            self.connection = self.__connect()
             self.info.clear()
+            self.connection = self.__connect()
             if self.__pool.dispatch.connect:
                 self.__pool.dispatch.connect(self.connection, self)
         return self.connection
@@ -539,7 +591,7 @@ class _ConnectionRecord(object):
     def __connect(self):
         try:
             self.starttime = time.time()
-            connection = self.__pool._creator()
+            connection = self.__pool._wrapped_creator(self)
             self.__pool.logger.debug("Created new connection %r", connection)
             return connection
         except Exception as e:
@@ -740,7 +792,7 @@ class _ConnectionFairy(object):
         """
         return self._connection_record.info
 
-    def invalidate(self, e=None):
+    def invalidate(self, e=None, soft=False):
         """Mark this connection as invalidated.
 
         This method can be called directly, and is also called as a result
@@ -749,6 +801,13 @@ class _ConnectionFairy(object):
         further use by the pool.  The invalidation mechanism proceeds
         via the :meth:`._ConnectionRecord.invalidate` internal method.
 
+        :param e: an exception object indicating a reason for the invalidation.
+
+        :param soft: if True, the connection isn't closed; instead, this
+         connection will be recycled on next checkout.
+
+         .. versionadded:: 1.0.3
+
         .. seealso::
 
             :ref:`pool_connection_invalidation`
@@ -759,9 +818,10 @@ class _ConnectionFairy(object):
             util.warn("Can't invalidate an already-closed connection.")
             return
         if self._connection_record:
-            self._connection_record.invalidate(e=e)
-        self.connection = None
-        self._checkin()
+            self._connection_record.invalidate(e=e, soft=soft)
+        if not soft:
+            self.connection = None
+            self._checkin()
 
     def cursor(self, *args, **kwargs):
         """Return a new DBAPI cursor for the underlying connection.