- Added connection pool events :meth:`ConnectionEvents.close`,

:meth:`.ConnectionEvents.detach`,
:meth:`.ConnectionEvents.close_detached`.

4 files changed, 137 insertions, 22 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index 0a5dc3ea0..97d1e3025 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -22,6 +22,13 @@
     :version: 1.1.0b1
 
     .. change::
+        :tags: feature, engine
+
+        Added connection pool events :meth:`ConnectionEvents.close`,
+        :meth:`.ConnectionEvents.detach`,
+        :meth:`.ConnectionEvents.close_detached`.
+
+    .. change::
         :tags: bug, orm, mysql
         :tickets: 3680
 
diff --git a/lib/sqlalchemy/events.py b/lib/sqlalchemy/events.py
index 1abef26d6..c679db37d 100644
--- a/lib/sqlalchemy/events.py
+++ b/lib/sqlalchemy/events.py
@@ -409,6 +409,46 @@ class PoolEvents(event.Events):
 
         """
 
+    def close(self, dbapi_connection, connection_record):
+        """Called when a DBAPI connection is closed.
+
+        The event is emitted before the close occurs.
+
+        The close of a connection can fail; typically this is because
+        the connection is already closed.  If the close operation fails,
+        the connection is discarded.
+
+        The :meth:`.close` event corresponds to a connection that's still
+        associated with the pool. To intercept close events for detached
+        connections use :meth:`.close_detached`.
+
+        .. versionadded:: 1.1
+
+        """
+
+    def detach(self, dbapi_connection, connection_record):
+        """Called when a DBAPI connection is "detached" from a pool.
+
+        This event is emitted after the detach occurs.  The connection
+        is no longer associated with the given connection record.
+
+        .. versionadded:: 1.1
+
+        """
+
+    def close_detached(self, dbapi_connection):
+        """Called when a detached DBAPI connection is closed.
+
+        The event is emitted before the close occurs.
+
+        The close of a connection can fail; typically this is because
+        the connection is already closed.  If the close operation fails,
+        the connection is discarded.
+
+        .. versionadded:: 1.1
+
+        """
+
 
 class ConnectionEvents(event.Events):
     """Available events for :class:`.Connectable`, which includes
diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 32b4736fa..fd1fefd00 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -286,6 +286,7 @@ class Pool(log.Identified):
 
     def _close_connection(self, connection):
         self.logger.debug("Closing connection %r", connection)
+
         try:
             self._dialect.do_close(connection)
         except Exception:
@@ -446,14 +447,9 @@ class _ConnectionRecord(object):
 
     def __init__(self, pool):
         self.__pool = pool
-        self.connection = self.__connect()
+        self.__connect(first_connect_check=True)
         self.finalize_callback = deque()
 
-        pool.dispatch.first_connect.\
-            for_modify(pool.dispatch).\
-            exec_once(self.connection, self)
-        pool.dispatch.connect(self.connection, self)
-
     connection = None
     """A reference to the actual DBAPI connection being tracked.
 
@@ -561,8 +557,6 @@ class _ConnectionRecord(object):
         if self.connection is None:
             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 \
                 time.time() - self.starttime > self.__pool._recycle:
             self.__pool.logger.info(
@@ -588,28 +582,36 @@ class _ConnectionRecord(object):
             self.__close()
             self.info.clear()
 
-            # ensure that if self.__connect() fails,
-            # we are not referring to the previous stale connection here
-            self.connection = None
             self.connection = self.__connect()
-
-            if self.__pool.dispatch.connect:
-                self.__pool.dispatch.connect(self.connection, self)
         return self.connection
 
     def __close(self):
         self.finalize_callback.clear()
+        if self.__pool.dispatch.close:
+            self.__pool.dispatch.close(self.connection, self)
         self.__pool._close_connection(self.connection)
 
-    def __connect(self):
+    def __connect(self, first_connect_check=False):
+        pool = self.__pool
+
+        # ensure any existing connection is removed, so that if
+        # creator fails, this attribute stays None
+        self.connection = None
         try:
             self.starttime = time.time()
-            connection = self.__pool._invoke_creator(self)
-            self.__pool.logger.debug("Created new connection %r", connection)
-            return connection
+            connection = pool._invoke_creator(self)
+            pool.logger.debug("Created new connection %r", connection)
+            self.connection = connection
         except Exception as e:
-            self.__pool.logger.debug("Error on connect(): %s", e)
+            pool.logger.debug("Error on connect(): %s", e)
             raise
+        else:
+            if first_connect_check:
+                pool.dispatch.first_connect.\
+                    for_modify(pool.dispatch).\
+                    exec_once(self.connection, self)
+            if pool.dispatch.connect:
+                pool.dispatch.connect(self.connection, self)
 
 
 def _finalize_fairy(connection, connection_record,
@@ -637,6 +639,8 @@ def _finalize_fairy(connection, connection_record,
 
             # Immediately close detached instances
             if not connection_record:
+                if pool.dispatch.close_detached:
+                    pool.dispatch.close_detached(connection)
                 pool._close_connection(connection)
         except BaseException as e:
             pool.logger.error(
@@ -868,14 +872,18 @@ class _ConnectionFairy(object):
         """
 
         if self._connection_record is not None:
-            _refs.remove(self._connection_record)
-            self._connection_record.fairy_ref = None
-            self._connection_record.connection = None
+            rec = self._connection_record
+            _refs.remove(rec)
+            rec.fairy_ref = None
+            rec.connection = None
             # TODO: should this be _return_conn?
             self._pool._do_return_conn(self._connection_record)
             self.info = self.info.copy()
             self._connection_record = None
 
+            if self._pool.dispatch.detach:
+                self._pool.dispatch.detach(self.connection, rec)
+
     def close(self):
         self._counter -= 1
         if self._counter == 0:
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 8551e1fcb..4547984ab 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -345,6 +345,66 @@ class PoolEventsTest(PoolTestBase):
 
         return p, canary
 
+    def _close_event_fixture(self):
+        p = self._queuepool_fixture()
+        canary = Mock()
+        event.listen(p, 'close', canary)
+
+        return p, canary
+
+    def _detach_event_fixture(self):
+        p = self._queuepool_fixture()
+        canary = Mock()
+        event.listen(p, 'detach', canary)
+
+        return p, canary
+
+    def _close_detached_event_fixture(self):
+        p = self._queuepool_fixture()
+        canary = Mock()
+        event.listen(p, 'close_detached', canary)
+
+        return p, canary
+
+    def test_close(self):
+        p, canary = self._close_event_fixture()
+
+        c1 = p.connect()
+
+        connection = c1.connection
+        rec = c1._connection_record
+
+        c1.close()
+
+        eq_(canary.mock_calls, [])
+
+        p.dispose()
+        eq_(canary.mock_calls, [call(connection, rec)])
+
+    def test_detach(self):
+        p, canary = self._detach_event_fixture()
+
+        c1 = p.connect()
+
+        connection = c1.connection
+        rec = c1._connection_record
+
+        c1.detach()
+
+        eq_(canary.mock_calls, [call(connection, rec)])
+
+    def test_detach_close(self):
+        p, canary = self._close_detached_event_fixture()
+
+        c1 = p.connect()
+
+        connection = c1.connection
+
+        c1.detach()
+
+        c1.close()
+        eq_(canary.mock_calls, [call(connection)])
+
     def test_first_connect_event(self):
         p, canary = self._first_connect_event_fixture()