- 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, 96 insertions, 2 deletions

diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index ff45b2d51..3d93cda89 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -4,11 +4,11 @@ from sqlalchemy import pool, select, event
 import sqlalchemy as tsa
 from sqlalchemy import testing
 from sqlalchemy.testing.util import gc_collect, lazy_gc
-from sqlalchemy.testing import eq_, assert_raises, is_not_
+from sqlalchemy.testing import eq_, assert_raises, is_not_, is_
 from sqlalchemy.testing.engines import testing_engine
 from sqlalchemy.testing import fixtures
 import random
-from sqlalchemy.testing.mock import Mock, call
+from sqlalchemy.testing.mock import Mock, call, patch
 import weakref
 
 join_timeout = 10
@@ -335,6 +335,13 @@ class PoolEventsTest(PoolTestBase):
 
         return p, canary
 
+    def _soft_invalidate_event_fixture(self):
+        p = self._queuepool_fixture()
+        canary = Mock()
+        event.listen(p, 'soft_invalidate', canary)
+
+        return p, canary
+
     def test_first_connect_event(self):
         p, canary = self._first_connect_event_fixture()
 
@@ -438,6 +445,31 @@ class PoolEventsTest(PoolTestBase):
         c1.close()
         eq_(canary, ['reset'])
 
+    def test_soft_invalidate_event_no_exception(self):
+        p, canary = self._soft_invalidate_event_fixture()
+
+        c1 = p.connect()
+        c1.close()
+        assert not canary.called
+        c1 = p.connect()
+        dbapi_con = c1.connection
+        c1.invalidate(soft=True)
+        assert canary.call_args_list[0][0][0] is dbapi_con
+        assert canary.call_args_list[0][0][2] is None
+
+    def test_soft_invalidate_event_exception(self):
+        p, canary = self._soft_invalidate_event_fixture()
+
+        c1 = p.connect()
+        c1.close()
+        assert not canary.called
+        c1 = p.connect()
+        dbapi_con = c1.connection
+        exc = Exception("hi")
+        c1.invalidate(exc, soft=True)
+        assert canary.call_args_list[0][0][0] is dbapi_con
+        assert canary.call_args_list[0][0][2] is exc
+
     def test_invalidate_event_no_exception(self):
         p, canary = self._invalidate_event_fixture()
 
@@ -1130,6 +1162,44 @@ class QueuePoolTest(PoolTestBase):
 
         eq_(len(success), 12, "successes: %s" % success)
 
+    def test_connrec_invalidated_within_checkout_no_race(self):
+        """Test that a concurrent ConnectionRecord.invalidate() which
+        occurs after the ConnectionFairy has called _ConnectionRecord.checkout()
+        but before the ConnectionFairy tests "fairy.connection is None"
+        will not result in an InvalidRequestError.
+
+        This use case assumes that a listener on the checkout() event
+        will be raising DisconnectionError so that a reconnect attempt
+        may occur.
+
+        """
+        dbapi = MockDBAPI()
+
+        def creator():
+            return dbapi.connect()
+
+        p = pool.QueuePool(creator=creator, pool_size=1, max_overflow=0)
+
+        conn = p.connect()
+        conn.close()
+
+        _existing_checkout = pool._ConnectionRecord.checkout
+
+        @classmethod
+        def _decorate_existing_checkout(cls, *arg, **kw):
+            fairy = _existing_checkout(*arg, **kw)
+            connrec = fairy._connection_record
+            connrec.invalidate()
+            return fairy
+
+        with patch(
+                "sqlalchemy.pool._ConnectionRecord.checkout",
+                _decorate_existing_checkout):
+            conn = p.connect()
+            is_(conn._connection_record.connection, None)
+        conn.close()
+
+
     @testing.requires.threading_with_mock
     @testing.requires.timing_intensive
     def test_notify_waiters(self):
@@ -1323,12 +1393,36 @@ class QueuePoolTest(PoolTestBase):
         c2 = p.connect()
         assert id(c2.connection) == c_id
 
+        c2_rec = c2._connection_record
         p._invalidate(c2)
+        assert c2_rec.connection is None
         c2.close()
         time.sleep(.5)
         c3 = p.connect()
         assert id(c3.connection) != c_id
 
+    @testing.requires.timing_intensive
+    def test_recycle_on_soft_invalidate(self):
+        p = self._queuepool_fixture(pool_size=1,
+                           max_overflow=0)
+        c1 = p.connect()
+        c_id = id(c1.connection)
+        c1.close()
+        c2 = p.connect()
+        assert id(c2.connection) == c_id
+
+        c2_rec = c2._connection_record
+        c2.invalidate(soft=True)
+        assert c2_rec.connection is c2.connection
+
+        c2.close()
+        time.sleep(.5)
+        c3 = p.connect()
+        assert id(c3.connection) != c_id
+        assert c3._connection_record is c2_rec
+        assert c2_rec.connection is c3.connection
+
+
     def _assert_cleanup_on_pooled_reconnect(self, dbapi, p):
         # p is QueuePool with size=1, max_overflow=2,
         # and one connection in the pool that will need to