Support pool.connect() event firing before all else

Fixed regression where a connection pool event specified with a keyword,
most notably ``insert=True``, would be lost when the event were set up.
This would prevent startup events that need to fire before dialect-level
events from working correctly.

The internal mechanics of the engine connection routine has been altered
such that it's now guaranteed that a user-defined event handler for the
:meth:`_pool.PoolEvents.connect` handler, when established using
``insert=True``, will allow an event handler to run that is definitely
invoked **before** any dialect-specific initialization starts up, most
notably when it does things like detect default schema name.
Previously, this would occur in most cases but not unconditionally.
A new example is added to the schema documentation illustrating how to
establish the "default schema name" within an on-connect event
(upcoming as part of I882edd5bbe06ee5b4d0a9c148854a57b2bcd4741)

Addiional changes to support setting default schema name:

The Oracle dialect now uses
``select sys_context( 'userenv', 'current_schema' ) from dual`` to get
the default schema name, rather than ``SELECT USER FROM DUAL``, to
accommodate for changes to the session-local schema name under Oracle.

Added a read/write ``.autocommit`` attribute to the DBAPI-adaptation layer
for the asyncpg dialect.   This so that when working with DBAPI-specific
schemes that need to use "autocommit" directly with the DBAPI connection,
the same ``.autocommit`` attribute which works with both psycopg2 as well
as pg8000 is available.

Fixes: #5716
Fixes: #5708
Change-Id: I7dce56b4345ffc720e25e2aaccb7e42bb29e5671

1 files changed, 50 insertions, 0 deletions

diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index bfdd14845..5cea604d6 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -39,6 +39,7 @@ from sqlalchemy.engine import url
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
+from sqlalchemy.testing import is_false
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
 from sqlalchemy.testing.assertions import assert_raises
@@ -802,6 +803,55 @@ class MiscBackendTest(
             with engine.connect():
                 eq_(engine.dialect._backslash_escapes, expected)
 
+    def test_dbapi_autocommit_attribute(self):
+        """all the supported DBAPIs have an .autocommit attribute.  make
+        sure it works and preserves isolation level.
+
+        This is added in particular to support the asyncpg dialect that
+        has a DBAPI compatibility layer.
+
+        """
+
+        with testing.db.connect().execution_options(
+            isolation_level="SERIALIZABLE"
+        ) as conn:
+            dbapi_conn = conn.connection.connection
+
+            is_false(dbapi_conn.autocommit)
+
+            with conn.begin():
+
+                existing_isolation = conn.exec_driver_sql(
+                    "show transaction isolation level"
+                ).scalar()
+                eq_(existing_isolation.upper(), "SERIALIZABLE")
+
+                txid1 = conn.exec_driver_sql("select txid_current()").scalar()
+                txid2 = conn.exec_driver_sql("select txid_current()").scalar()
+                eq_(txid1, txid2)
+
+            dbapi_conn.autocommit = True
+
+            with conn.begin():
+                # magic way to see if we are in autocommit mode from
+                # the server's perspective
+                txid1 = conn.exec_driver_sql("select txid_current()").scalar()
+                txid2 = conn.exec_driver_sql("select txid_current()").scalar()
+                ne_(txid1, txid2)
+
+            dbapi_conn.autocommit = False
+
+            with conn.begin():
+
+                existing_isolation = conn.exec_driver_sql(
+                    "show transaction isolation level"
+                ).scalar()
+                eq_(existing_isolation.upper(), "SERIALIZABLE")
+
+                txid1 = conn.exec_driver_sql("select txid_current()").scalar()
+                txid2 = conn.exec_driver_sql("select txid_current()").scalar()
+                eq_(txid1, txid2)
+
     def test_readonly_flag_connection(self):
         with testing.db.connect() as conn:
             # asyncpg requires serializable for readonly..