fully support isolation_level parameter in base dialect

Generalized the :paramref:`_sa.create_engine.isolation_level` parameter to
the base dialect so that it is no longer dependent on individual dialects
to be present. This parameter sets up the "isolation level" setting to
occur for all new database connections as soon as they are created by the
connection pool, where the value then stays set without being reset on
every checkin.

The :paramref:`_sa.create_engine.isolation_level` parameter is essentially
equivalent in functionality to using the
:paramref:`_engine.Engine.execution_options.isolation_level` parameter via
:meth:`_engine.Engine.execution_options` for an engine-wide setting. The
difference is in that the former setting assigns the isolation level just
once when a connection is created, the latter sets and resets the given
level on each connection checkout.

Fixes: #6342
Change-Id: Id81d6b1c1a94371d901ada728a610696e09e9741

12 files changed, 53 insertions, 187 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 0077f7fa1..974cae4f7 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2725,7 +2725,6 @@ class MSDialect(default.DefaultDialect):
         query_timeout=None,
         use_scope_identity=True,
         schema_name="dbo",
-        isolation_level=None,
         deprecate_large_types=None,
         json_serializer=None,
         json_deserializer=None,
@@ -2748,7 +2747,6 @@ class MSDialect(default.DefaultDialect):
 
         super(MSDialect, self).__init__(**opts)
 
-        self.isolation_level = isolation_level
         self._json_serializer = json_serializer
         self._json_deserializer = json_deserializer
 
@@ -2771,14 +2769,10 @@ class MSDialect(default.DefaultDialect):
         ]
     )
 
+    def get_isolation_level_values(self, dbapi_conn):
+        return list(self._isolation_lookup)
+
     def set_isolation_level(self, connection, level):
-        level = level.replace("_", " ")
-        if level not in self._isolation_lookup:
-            raise exc.ArgumentError(
-                "Invalid value '%s' for isolation_level. "
-                "Valid isolation levels for %s are %s"
-                % (level, self.name, ", ".join(self._isolation_lookup))
-            )
         cursor = connection.cursor()
         cursor.execute("SET TRANSACTION ISOLATION LEVEL %s" % level)
         cursor.close()
@@ -2833,16 +2827,6 @@ class MSDialect(default.DefaultDialect):
         self._setup_version_attributes()
         self._setup_supports_nvarchar_max(connection)
 
-    def on_connect(self):
-        if self.isolation_level is not None:
-
-            def connect(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            return connect
-        else:
-            return None
-
     def _setup_version_attributes(self):
         if self.server_version_info[0] not in list(range(8, 17)):
             util.warn(
diff --git a/lib/sqlalchemy/dialects/mssql/pymssql.py b/lib/sqlalchemy/dialects/mssql/pymssql.py
index b559384ba..18bee1890 100644
--- a/lib/sqlalchemy/dialects/mssql/pymssql.py
+++ b/lib/sqlalchemy/dialects/mssql/pymssql.py
@@ -125,6 +125,9 @@ class MSDialect_pymssql(MSDialect):
         else:
             return False
 
+    def get_isolation_level_values(self, dbapi_conn):
+        return super().get_isolation_level_values(dbapi_conn) + ["AUTOCOMMIT"]
+
     def set_isolation_level(self, connection, level):
         if level == "AUTOCOMMIT":
             connection.autocommit(True)
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 5382e00db..af2d7bdbb 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2385,7 +2385,6 @@ class MySQLDialect(default.DefaultDialect):
 
     def __init__(
         self,
-        isolation_level=None,
         json_serializer=None,
         json_deserializer=None,
         is_mariadb=None,
@@ -2393,49 +2392,20 @@ class MySQLDialect(default.DefaultDialect):
     ):
         kwargs.pop("use_ansiquotes", None)  # legacy
         default.DefaultDialect.__init__(self, **kwargs)
-        self.isolation_level = isolation_level
         self._json_serializer = json_serializer
         self._json_deserializer = json_deserializer
         self._set_mariadb(is_mariadb, None)
 
-    def on_connect(self):
-        if self.isolation_level is not None:
-
-            def connect(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            return connect
-        else:
-            return None
-
-    _isolation_lookup = set(
-        [
+    def get_isolation_level_values(self, dbapi_conn):
+        return (
             "SERIALIZABLE",
             "READ UNCOMMITTED",
             "READ COMMITTED",
             "REPEATABLE READ",
-        ]
-    )
-
-    def set_isolation_level(self, connection, level):
-        level = level.replace("_", " ")
-
-        # adjust for ConnectionFairy being present
-        # allows attribute set e.g. "connection.autocommit = True"
-        # to work properly
-        if hasattr(connection, "dbapi_connection"):
-            connection = connection.dbapi_connection
-
-        self._set_isolation_level(connection, level)
+        )
 
-    def _set_isolation_level(self, connection, level):
-        if level not in self._isolation_lookup:
-            raise exc.ArgumentError(
-                "Invalid value '%s' for isolation_level. "
-                "Valid isolation levels for %s are %s"
-                % (level, self.name, ", ".join(self._isolation_lookup))
-            )
-        cursor = connection.cursor()
+    def set_isolation_level(self, dbapi_conn, level):
+        cursor = dbapi_conn.cursor()
         cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL %s" % level)
         cursor.execute("COMMIT")
         cursor.close()
diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index dfe719c28..1e57c779d 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -308,23 +308,22 @@ class MySQLDialect_mysqldb(MySQLDialect):
         else:
             return cset_name()
 
-    _isolation_lookup = set(
-        [
+    def get_isolation_level_values(self, dbapi_conn):
+        return (
             "SERIALIZABLE",
             "READ UNCOMMITTED",
             "READ COMMITTED",
             "REPEATABLE READ",
             "AUTOCOMMIT",
-        ]
-    )
+        )
 
-    def _set_isolation_level(self, connection, level):
+    def set_isolation_level(self, dbapi_conn, level):
         if level == "AUTOCOMMIT":
-            connection.autocommit(True)
+            dbapi_conn.autocommit(True)
         else:
-            connection.autocommit(False)
-            super(MySQLDialect_mysqldb, self)._set_isolation_level(
-                connection, level
+            dbapi_conn.autocommit(False)
+            super(MySQLDialect_mysqldb, self).set_isolation_level(
+                dbapi_conn, level
             )
 
 
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 229a54b95..9846a65bd 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1573,10 +1573,8 @@ class OracleDialect(default.DefaultDialect):
             # use the default
             return None
 
-    _isolation_lookup = ["READ COMMITTED", "SERIALIZABLE"]
-
-    def get_isolation_level(self, connection):
-        raise NotImplementedError("implemented by cx_Oracle dialect")
+    def get_isolation_level_values(self, dbapi_conn):
+        return ["READ COMMITTED", "SERIALIZABLE"]
 
     def get_default_isolation_level(self, dbapi_conn):
         try:
@@ -1586,9 +1584,6 @@ class OracleDialect(default.DefaultDialect):
         except:
             return "READ COMMITTED"
 
-    def set_isolation_level(self, connection, level):
-        raise NotImplementedError("implemented by cx_Oracle dialect")
-
     def has_table(self, connection, table_name, schema=None):
         self._ensure_has_table_connection(connection)
 
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index 23f619a12..2cfcb0e5c 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -1024,6 +1024,9 @@ class OracleDialect_cx_oracle(OracleDialect):
 
         return result
 
+    def get_isolation_level_values(self, dbapi_conn):
+        return super().get_isolation_level_values(dbapi_conn) + ["AUTOCOMMIT"]
+
     def set_isolation_level(self, connection, level):
         if hasattr(connection, "dbapi_connection"):
             dbapi_connection = connection.dbapi_connection
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 28374ed60..fe1f9fd5a 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -933,20 +933,11 @@ class PGDialect_asyncpg(PGDialect):
             "SERIALIZABLE": "serializable",
         }
 
-    def set_isolation_level(self, connection, level):
-        try:
-            level = self._isolation_lookup[level.replace("_", " ")]
-        except KeyError as err:
-            util.raise_(
-                exc.ArgumentError(
-                    "Invalid value '%s' for isolation_level. "
-                    "Valid isolation levels for %s are %s"
-                    % (level, self.name, ", ".join(self._isolation_lookup))
-                ),
-                replace_context=err,
-            )
+    def get_isolation_level_values(self, dbapi_conn):
+        return list(self._isolation_lookup)
 
-        connection.set_isolation_level(level)
+    def set_isolation_level(self, connection, level):
+        connection.set_isolation_level(self._isolation_lookup[level])
 
     def set_readonly(self, connection, value):
         connection.readonly = value
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index a00c26e87..7d86ccd01 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3148,7 +3148,6 @@ class PGDialect(default.DefaultDialect):
     preparer = PGIdentifierPreparer
     execution_ctx_cls = PGExecutionContext
     inspector = PGInspector
-    isolation_level = None
 
     implicit_returning = True
     full_returning = True
@@ -3195,19 +3194,9 @@ class PGDialect(default.DefaultDialect):
     _supports_create_index_concurrently = True
     _supports_drop_index_concurrently = True
 
-    def __init__(
-        self,
-        isolation_level=None,
-        json_serializer=None,
-        json_deserializer=None,
-        **kwargs
-    ):
+    def __init__(self, json_serializer=None, json_deserializer=None, **kwargs):
         default.DefaultDialect.__init__(self, **kwargs)
 
-        # the isolation_level parameter to the PGDialect itself is legacy.
-        # still works however the execution_options method is the one that
-        # is documented.
-        self.isolation_level = isolation_level
         self._json_deserializer = json_deserializer
         self._json_serializer = json_serializer
 
@@ -3247,33 +3236,17 @@ class PGDialect(default.DefaultDialect):
         )
         self.supports_identity_columns = self.server_version_info >= (10,)
 
-    def on_connect(self):
-        if self.isolation_level is not None:
-
-            def connect(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            return connect
-        else:
-            return None
-
-    _isolation_lookup = set(
-        [
+    def get_isolation_level_values(self, dbapi_conn):
+        # note the generic dialect doesn't have AUTOCOMMIT, however
+        # all postgresql dialects should include AUTOCOMMIT.
+        return (
             "SERIALIZABLE",
             "READ UNCOMMITTED",
             "READ COMMITTED",
             "REPEATABLE READ",
-        ]
-    )
+        )
 
     def set_isolation_level(self, connection, level):
-        level = level.replace("_", " ")
-        if level not in self._isolation_lookup:
-            raise exc.ArgumentError(
-                "Invalid value '%s' for isolation_level. "
-                "Valid isolation levels for %s are %s"
-                % (level, self.name, ", ".join(self._isolation_lookup))
-            )
         cursor = connection.cursor()
         cursor.execute(
             "SET SESSION CHARACTERISTICS AS TRANSACTION "
diff --git a/lib/sqlalchemy/dialects/postgresql/pg8000.py b/lib/sqlalchemy/dialects/postgresql/pg8000.py
index a94f9dcdb..324007e7e 100644
--- a/lib/sqlalchemy/dialects/postgresql/pg8000.py
+++ b/lib/sqlalchemy/dialects/postgresql/pg8000.py
@@ -437,6 +437,15 @@ class PGDialect_pg8000(PGDialect):
         # connection was closed normally
         return "connection is closed" in str(e)
 
+    def get_isolation_level_values(self, dbapi_conn):
+        return (
+            "AUTOCOMMIT",
+            "READ COMMITTED",
+            "READ UNCOMMITTED",
+            "REPEATABLE READ",
+            "SERIALIZABLE",
+        )
+
     def set_isolation_level(self, connection, level):
         level = level.replace("_", " ")
 
@@ -446,7 +455,7 @@ class PGDialect_pg8000(PGDialect):
 
         if level == "AUTOCOMMIT":
             connection.autocommit = True
-        elif level in self._isolation_lookup:
+        else:
             connection.autocommit = False
             cursor = connection.cursor()
             cursor.execute(
@@ -455,12 +464,6 @@ class PGDialect_pg8000(PGDialect):
             )
             cursor.execute("COMMIT")
             cursor.close()
-        else:
-            raise exc.ArgumentError(
-                "Invalid value '%s' for isolation_level. "
-                "Valid isolation levels for %s are %s or AUTOCOMMIT"
-                % (level, self.name, ", ".join(self._isolation_lookup))
-            )
 
     def set_readonly(self, connection, value):
         cursor = connection.cursor()
@@ -562,13 +565,6 @@ class PGDialect_pg8000(PGDialect):
 
             fns.append(on_connect)
 
-        if self.isolation_level is not None:
-
-            def on_connect(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            fns.append(on_connect)
-
         if self._json_deserializer:
 
             def on_connect(conn):
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index aadd11059..f62830a0d 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -741,20 +741,11 @@ class PGDialect_psycopg2(PGDialect):
             "SERIALIZABLE": extensions.ISOLATION_LEVEL_SERIALIZABLE,
         }
 
-    def set_isolation_level(self, connection, level):
-        try:
-            level = self._isolation_lookup[level.replace("_", " ")]
-        except KeyError as err:
-            util.raise_(
-                exc.ArgumentError(
-                    "Invalid value '%s' for isolation_level. "
-                    "Valid isolation levels for %s are %s"
-                    % (level, self.name, ", ".join(self._isolation_lookup))
-                ),
-                replace_context=err,
-            )
+    def get_isolation_level_values(self, dbapi_conn):
+        return list(self._isolation_lookup)
 
-        connection.set_isolation_level(level)
+    def set_isolation_level(self, connection, level):
+        connection.set_isolation_level(self._isolation_lookup[level])
 
     def set_readonly(self, connection, value):
         connection.readonly = value
@@ -798,13 +789,6 @@ class PGDialect_psycopg2(PGDialect):
 
             fns.append(on_connect)
 
-        if self.isolation_level is not None:
-
-            def on_connect(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            fns.append(on_connect)
-
         if self.dbapi and self.use_native_uuid:
 
             def on_connect(conn):
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index dc8425859..6c22c8ef3 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1815,7 +1815,6 @@ class SQLiteDialect(default.DefaultDialect):
     preparer = SQLiteIdentifierPreparer
     ischema_names = ischema_names
     colspecs = colspecs
-    isolation_level = None
 
     construct_arguments = [
         (
@@ -1856,7 +1855,6 @@ class SQLiteDialect(default.DefaultDialect):
     )
     def __init__(
         self,
-        isolation_level=None,
         native_datetime=False,
         json_serializer=None,
         json_deserializer=None,
@@ -1865,7 +1863,6 @@ class SQLiteDialect(default.DefaultDialect):
         **kwargs
     ):
         default.DefaultDialect.__init__(self, **kwargs)
-        self.isolation_level = isolation_level
 
         if _json_serializer:
             json_serializer = _json_serializer
@@ -1918,22 +1915,12 @@ class SQLiteDialect(default.DefaultDialect):
         {"READ UNCOMMITTED": 1, "SERIALIZABLE": 0}
     )
 
+    def get_isolation_level_values(self, dbapi_conn):
+        return list(self._isolation_lookup)
+
     def set_isolation_level(self, connection, level):
-        try:
-            isolation_level = self._isolation_lookup[level.replace("_", " ")]
-        except KeyError as err:
-            util.raise_(
-                exc.ArgumentError(
-                    "Invalid value '%s' for isolation_level. "
-                    "Valid isolation levels for %s are %s"
-                    % (
-                        level,
-                        self.name,
-                        ", ".join(self._isolation_lookup),
-                    )
-                ),
-                replace_context=err,
-            )
+        isolation_level = self._isolation_lookup[level]
+
         cursor = connection.cursor()
         cursor.execute("PRAGMA read_uncommitted = %d" % isolation_level)
         cursor.close()
@@ -1960,16 +1947,6 @@ class SQLiteDialect(default.DefaultDialect):
         else:
             assert False, "Unknown isolation level %s" % value
 
-    def on_connect(self):
-        if self.isolation_level is not None:
-
-            def connect(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            return connect
-        else:
-            return None
-
     @reflection.cache
     def get_schema_names(self, connection, **kw):
         s = "PRAGMA database_list"
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlite.py b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
index 10912e0d5..45a35be65 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
@@ -504,8 +504,6 @@ class SQLiteDialect_pysqlite(SQLiteDialect):
             )
 
     def on_connect(self):
-        connect = super(SQLiteDialect_pysqlite, self).on_connect()
-
         def regexp(a, b):
             if b is None:
                 return None
@@ -524,13 +522,6 @@ class SQLiteDialect_pysqlite(SQLiteDialect):
 
         fns = [set_regexp]
 
-        if self.isolation_level is not None:
-
-            def iso_level(conn):
-                self.set_isolation_level(conn, self.isolation_level)
-
-            fns.append(iso_level)
-
         def connect(conn):
             for fn in fns:
                 fn(conn)