Don't link on_connect to first_connect event handler

Adjusted the dialect initialization process such that the
:meth:`_engine.Dialect.on_connect` is not called a second time on the first
connection.   The hook is called first, then the
:meth:`_engine.Dialect.initialize` is called if that connection is the
first for that dialect, then no more events are called.   This eliminates
the two calls to the "on_connect" function which can produce very difficult
debugging situations.

Fixes: #5497
Change-Id: Icefc2e884e30ee7b4ac84b99dc54bf992a6085e3

6 files changed, 106 insertions, 13 deletions

diff --git a/doc/build/changelog/unreleased_14/5497.rst b/doc/build/changelog/unreleased_14/5497.rst
new file mode 100644
index 000000000..0f414a379
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/5497.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: engine, change
+    :tickets: 5497
+
+    Adjusted the dialect initialization process such that the
+    :meth:`_engine.Dialect.on_connect` is not called a second time
+    on the first connection.   The hook is called first, then the
+    :meth:`_engine.Dialect.initialize` is called if that connection is the
+    first for that dialect, then no more events are called.   This eliminates
+    the two calls to the "on_connect" function which can produce very
+    difficult debugging situations.
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index 985a12fa0..dc895ee15 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -627,9 +627,9 @@ def create_engine(url, **kwargs):
                 )
                 if conn is None:
                     return
+
                 do_on_connect(conn)
 
-            event.listen(pool, "first_connect", on_connect)
             event.listen(pool, "connect", on_connect)
 
         def first_connect(dbapi_connection, connection_record):
@@ -640,9 +640,17 @@ def create_engine(url, **kwargs):
             dialect.initialize(c)
             dialect.do_rollback(c.connection)
 
-        event.listen(
-            pool, "first_connect", first_connect, _once_unless_exception=True
-        )
+        if do_on_connect:
+            event.listen(
+                pool, "connect", first_connect, _once_unless_exception=True
+            )
+        else:
+            event.listen(
+                pool,
+                "first_connect",
+                first_connect,
+                _once_unless_exception=True,
+            )
 
     dialect_cls.engine_created(engine)
     if entrypoint is not dialect_cls:
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 8d3c5de15..c76f820f9 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -120,6 +120,8 @@ class DefaultDialect(interfaces.Dialect):
     max_identifier_length = 9999
     _user_defined_max_identifier_length = None
 
+    isolation_level = None
+
     # length at which to truncate
     # the name of an index.
     # Usually None to indicate
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 1baef29af..7d51ab159 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -208,6 +208,9 @@ class Dialect(object):
         The initialize() method of the base dialect should be called via
         super().
 
+        .. note:: as of SQLAlchemy 1.4, this method is called **before**
+           any :meth:`_engine.Dialect.on_connect` hooks are called.
+
         """
 
         pass
@@ -745,16 +748,13 @@ class Dialect(object):
         isolation modes, Unicode modes, etc.
 
         The "do_on_connect" callable is invoked by using the
-        :meth:`_events.PoolEvents.first_connect` and
         :meth:`_events.PoolEvents.connect` event
-        hooks, then unwrapping the DBAPI connection and passing it into the
-        callable.  The reason it is invoked for both events is so that any
-        dialect-level initialization that occurs upon first connection, which
-        also makes use of the :meth:`_events.PoolEvents.first_connect` method,
-        will
-        proceed after this hook has been called. This currently means the
-        hook is in fact called twice for the very first  connection in which a
-        dialect creates; and once per connection afterwards.
+        hook, then unwrapping the DBAPI connection and passing it into the
+        callable.
+
+        .. versionchanged:: 1.4 the on_connect hook is no longer called twice
+           for the first connection of a dialect.  The on_connect hook is still
+           called before the :meth:`_engine.Dialect.initialize` method however.
 
         If None is returned, no event listener is generated.
 
diff --git a/lib/sqlalchemy/testing/engines.py b/lib/sqlalchemy/testing/engines.py
index 280e6901e..bb137cb32 100644
--- a/lib/sqlalchemy/testing/engines.py
+++ b/lib/sqlalchemy/testing/engines.py
@@ -336,6 +336,9 @@ class DBAPIProxyCursor(object):
     def executemany(self, stmt, params, **kw):
         return self.cursor.executemany(stmt, params, **kw)
 
+    def __iter__(self):
+        return iter(self.cursor)
+
     def __getattr__(self, key):
         return getattr(self.cursor, key)
 
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index ec255ba04..89d5c6348 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -26,6 +26,7 @@ from sqlalchemy import VARCHAR
 from sqlalchemy.engine import default
 from sqlalchemy.engine.base import Connection
 from sqlalchemy.engine.base import Engine
+from sqlalchemy.pool import QueuePool
 from sqlalchemy.sql import column
 from sqlalchemy.sql import literal
 from sqlalchemy.testing import assert_raises
@@ -2748,6 +2749,74 @@ class HandleInvalidatedOnConnectTest(fixtures.TestBase):
         except tsa.exc.DBAPIError as de:
             assert de.connection_invalidated
 
+    @testing.only_on("sqlite+pysqlite")
+    def test_initialize_connect_calls(self):
+        """test for :ticket:`5497`, on_connect not called twice"""
+
+        m1 = Mock()
+        cls_ = testing.db.dialect.__class__
+
+        class SomeDialect(cls_):
+            def initialize(self, connection):
+                super(SomeDialect, self).initialize(connection)
+                m1.initialize(connection)
+
+            def on_connect(self):
+                oc = super(SomeDialect, self).on_connect()
+
+                def my_on_connect(conn):
+                    if oc:
+                        oc(conn)
+                    m1.on_connect(conn)
+
+                return my_on_connect
+
+        u1 = Mock(
+            username=None,
+            password=None,
+            host=None,
+            port=None,
+            query={},
+            database=None,
+            _instantiate_plugins=lambda kw: [],
+            _get_entrypoint=Mock(
+                return_value=Mock(get_dialect_cls=lambda u: SomeDialect)
+            ),
+        )
+        eng = create_engine(u1, poolclass=QueuePool)
+        eq_(
+            eng.name, "sqlite"
+        )  # make sure other dialects aren't getting pulled in here
+        c = eng.connect()
+        dbapi_conn_one = c.connection.connection
+        c.close()
+
+        eq_(
+            m1.mock_calls,
+            [call.on_connect(dbapi_conn_one), call.initialize(mock.ANY)],
+        )
+
+        c = eng.connect()
+
+        eq_(
+            m1.mock_calls,
+            [call.on_connect(dbapi_conn_one), call.initialize(mock.ANY)],
+        )
+
+        c2 = eng.connect()
+        dbapi_conn_two = c2.connection.connection
+
+        is_not_(dbapi_conn_one, dbapi_conn_two)
+
+        eq_(
+            m1.mock_calls,
+            [
+                call.on_connect(dbapi_conn_one),
+                call.initialize(mock.ANY),
+                call.on_connect(dbapi_conn_two),
+            ],
+        )
+
 
 class DialectEventTest(fixtures.TestBase):
     @contextmanager