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

26 files changed, 437 insertions, 326 deletions

diff --git a/doc/build/changelog/unreleased_20/6342.rst b/doc/build/changelog/unreleased_20/6342.rst
new file mode 100644
index 000000000..9283c1937
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/6342.rst
@@ -0,0 +1,18 @@
+.. change::
+    :tags: engine, usecase
+    :tickets: 6342
+
+    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.
diff --git a/doc/build/core/connections.rst b/doc/build/core/connections.rst
index 0cf3c2e2a..08e782351 100644
--- a/doc/build/core/connections.rst
+++ b/doc/build/core/connections.rst
@@ -284,11 +284,17 @@ that loses not only "read committed" but also loses atomicity.
   not change autocommit mode**).
 
 SQLAlchemy dialects should support these isolation levels as well as autocommit
-to as great a degree as possible.   The levels are set via family of
-"execution_options" parameters and methods that are throughout the Core, such
-as the :meth:`_engine.Connection.execution_options` method.   The parameter is
-known as :paramref:`_engine.Connection.execution_options.isolation_level` and
-the values are strings which are typically a subset of the following names::
+to as great a degree as possible.
+
+Setting Isolation Level or DBAPI Autocommit for a Connection
+------------------------------------------------------------
+
+For an individual :class:`_engine.Connection` object that's acquired from
+:meth:`.Engine.connect`, the isolation level can be set for the duration of
+that :class:`_engine.Connection` object using the
+:meth:`_engine.Connection.execution_options` method. The parameter is known as
+:paramref:`_engine.Connection.execution_options.isolation_level` and the values
+are strings which are typically a subset of the following names::
 
     # possible values for Connection.execution_options(isolation_level="<value>")
 
@@ -319,11 +325,43 @@ begin a transaction::
    objects with different execution options, which nonetheless share the same
    dialect and connection pool.
 
+.. note:: The :paramref:`_engine.Connection.execution_options.isolation_level`
+   parameter necessarily does not apply to statement level options, such as
+   that of :meth:`_sql.Executable.execution_options`, and will be rejected if
+   set at this level. This because the option must be set on a DBAPI connection
+   on a per-transaction basis.
+
+Setting Isolation Level or DBAPI Autocommit for an Engine
+----------------------------------------------------------
+
 The :paramref:`_engine.Connection.execution_options.isolation_level` option may
-also be set engine wide, as is often preferable.  This is achieved by
-passing it within the :paramref:`_sa.create_engine.execution_options`
-parameter to :func:`_sa.create_engine`::
+also be set engine wide, as is often preferable.  This may be
+achieved by passing the :paramref:`_sa.create_engine.isolation_level`
+parameter to :func:`.sa.create_engine`::
+
+    from sqlalchemy import create_engine
 
+    eng = create_engine(
+        "postgresql://scott:tiger@localhost/test",
+        isolation_level="REPEATABLE READ"
+    )
+
+With the above setting, each new DBAPI connection the moment it's created will
+be set to use a ``"REPEATABLE READ"`` isolation level setting for all
+subsequent operations.
+
+.. _dbapi_autocommit_multiple:
+
+Maintaining Multiple Isolation Levels for a Single Engine
+----------------------------------------------------------
+
+The isolation level may also be set per engine, with a potentially greater
+level of flexibility, using either the
+:paramref:`_sa.create_engine.execution_options` parameter to
+:func:`_sa.create_engine` or the :meth:`_engine.Engine.execution_options`
+method, the latter of which will create a copy of the :class:`.Engine` that
+shares the dialect and connection pool of the original engine, but has its own
+per-connection isolation level setting::
 
     from sqlalchemy import create_engine
 
@@ -336,15 +374,18 @@ parameter to :func:`_sa.create_engine`::
 
 With the above setting, the DBAPI connection will be set to use a
 ``"REPEATABLE READ"`` isolation level setting for each new transaction
-begun.
+begun; but the connection as pooled will be reset to the original isolation
+level that was present when the connection first occurred.   At the level
+of :func:`_sa.create_engine`, the end effect is not any different
+from using the :paramref:`_sa.create_engine.isolation_level` parameter.
 
-An application that frequently chooses to run operations within different
-isolation levels may wish to create multiple "sub-engines" of a lead
+However, an application that frequently chooses to run operations within
+different isolation levels may wish to create multiple "sub-engines" of a lead
 :class:`_engine.Engine`, each of which will be configured to a different
-isolation level.  One such use case is an application that has operations
-that break into "transactional" and "read-only" operations, a separate
-:class:`_engine.Engine` that makes use of ``"AUTOCOMMIT"`` may be
-separated off from the main engine::
+isolation level. One such use case is an application that has operations that
+break into "transactional" and "read-only" operations, a separate
+:class:`_engine.Engine` that makes use of ``"AUTOCOMMIT"`` may be separated off
+from the main engine::
 
     from sqlalchemy import create_engine
 
@@ -352,7 +393,6 @@ separated off from the main engine::
 
     autocommit_engine = eng.execution_options(isolation_level="AUTOCOMMIT")
 
-
 Above, the :meth:`_engine.Engine.execution_options` method creates a shallow
 copy of the original :class:`_engine.Engine`.  Both ``eng`` and
 ``autocommit_engine`` share the same dialect and connection pool.  However, the
@@ -363,11 +403,6 @@ The isolation level setting, regardless of which one it is, is unconditionally
 reverted when a connection is returned to the connection pool.
 
 
-.. note:: The :paramref:`_engine.Connection.execution_options.isolation_level`
-   parameter necessarily does not apply to statement level options, such as
-   that of :meth:`_sql.Executable.execution_options`.  This because the option
-   must be set on a DBAPI connection on a per-transaction basis.
-
 .. seealso::
 
       :ref:`SQLite Transaction Isolation <sqlite_isolation_level>`
@@ -378,6 +413,8 @@ reverted when a connection is returned to the connection pool.
 
       :ref:`SQL Server Transaction Isolation <mssql_isolation_level>`
 
+      :ref:`Oracle Transaction Isolation <oracle_isolation_level>`
+
       :ref:`session_transaction_isolation` - for the ORM
 
       :ref:`faq_execute_retry_autocommit` - a recipe that uses DBAPI autocommit
diff --git a/lib/sqlalchemy/connectors/pyodbc.py b/lib/sqlalchemy/connectors/pyodbc.py
index 9661015ad..411985b5d 100644
--- a/lib/sqlalchemy/connectors/pyodbc.py
+++ b/lib/sqlalchemy/connectors/pyodbc.py
@@ -172,6 +172,9 @@ class PyODBCConnector(Connector):
             ]
         )
 
+    def get_isolation_level_values(self, dbapi_conn):
+        return super().get_isolation_level_values(dbapi_conn) + ["AUTOCOMMIT"]
+
     def set_isolation_level(self, connection, level):
         # adjust for ConnectionFairy being present
         # allows attribute set e.g. "connection.autocommit = True"
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)
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index ef6282525..24f8a8a87 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -2392,9 +2392,6 @@ class Engine(ConnectionEventsTarget, log.Identified):
         * The logging configuration and logging_name is copied from the parent
           :class:`_engine.Engine`.
 
-        .. TODO: the below autocommit link will have a more specific ref
-           for the example in an upcoming commit
-
         The intent of the :meth:`_engine.Engine.execution_options` method is
         to implement schemes where multiple :class:`_engine.Engine`
         objects refer to the same connection pool, but are differentiated
@@ -2404,7 +2401,7 @@ class Engine(ConnectionEventsTarget, log.Identified):
         :class:`_engine.Engine`
         has a lower :term:`isolation level` setting configured or is even
         transaction-disabled using "autocommit".  An example of this
-        configuration is at :ref:`dbapi_autocommit`.
+        configuration is at :ref:`dbapi_autocommit_multiple`.
 
         Another example is one that
         uses a custom option ``shard_id`` which is consumed by an event
diff --git a/lib/sqlalchemy/engine/characteristics.py b/lib/sqlalchemy/engine/characteristics.py
index c00bff40d..2543f591b 100644
--- a/lib/sqlalchemy/engine/characteristics.py
+++ b/lib/sqlalchemy/engine/characteristics.py
@@ -50,7 +50,7 @@ class IsolationLevelCharacteristic(ConnectionCharacteristic):
         dialect.reset_isolation_level(dbapi_conn)
 
     def set_characteristic(self, dialect, dbapi_conn, value):
-        dialect.set_isolation_level(dbapi_conn, value)
+        dialect._assert_and_set_isolation_level(dbapi_conn, value)
 
     def get_characteristic(self, dialect, dbapi_conn):
         return dialect.get_isolation_level(dbapi_conn)
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index e6da1d8e6..5932bfb9b 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -201,34 +201,32 @@ def create_engine(url, **kwargs):
            should **always be set to True**.  Some SQLAlchemy features will
            fail to function properly if this flag is set to ``False``.
 
-    :param isolation_level: this string parameter is interpreted by various
-        dialects in order to affect the transaction isolation level of the
-        database connection.   The parameter essentially accepts some subset of
-        these string arguments: ``"SERIALIZABLE"``, ``"REPEATABLE READ"``,
-        ``"READ COMMITTED"``, ``"READ UNCOMMITTED"`` and ``"AUTOCOMMIT"``.
-        Behavior here varies per backend, and
-        individual dialects should be consulted directly.
-
-        Note that the isolation level can also be set on a
-        per-:class:`_engine.Connection` basis as well, using the
+    :param isolation_level: optional string name of an isolation level
+        which will be set on all new connections unconditionally.
+        Isolation levels are typically some subset of the string names
+        ``"SERIALIZABLE"``, ``"REPEATABLE READ"``,
+        ``"READ COMMITTED"``, ``"READ UNCOMMITTED"`` and ``"AUTOCOMMIT"``
+        based on backend.
+
+        The :paramref:`_sa.create_engine.isolation_level` parameter is
+        in contrast to the
         :paramref:`.Connection.execution_options.isolation_level`
-        feature.
+        execution option, which may be set on an individual
+        :class:`.Connection`, as well as the same parameter passed to
+        :meth:`.Engine.execution_options`, where it may be used to create
+        multiple engines with different isolation levels that share a common
+        connection pool and dialect.
+
+        .. versionchanged:: 2.0 The
+           :paramref:`_sa.create_engine.isolation_level`
+           parameter has been generalized to work on all dialects which support
+           the concept of isolation level, and is provided as a more succinct,
+           up front configuration switch in contrast to the execution option
+           which is more of an ad-hoc programmatic option.
 
         .. seealso::
 
-            :attr:`_engine.Connection.default_isolation_level`
-            - view default level
-
-            :paramref:`.Connection.execution_options.isolation_level`
-            - set per :class:`_engine.Connection` isolation level
-
-            :ref:`SQLite Transaction Isolation <sqlite_isolation_level>`
-
-            :ref:`PostgreSQL Transaction Isolation <postgresql_isolation_level>`
-
-            :ref:`MySQL Transaction Isolation <mysql_isolation_level>`
-
-            :ref:`session_transaction_isolation` - for the ORM
+            :ref:`dbapi_autocommit`
 
     :param json_deserializer: for dialects that support the
         :class:`_types.JSON`
@@ -594,6 +592,10 @@ def create_engine(url, **kwargs):
 
             event.listen(pool, "connect", on_connect)
 
+        builtin_on_connect = dialect._builtin_onconnect()
+        if builtin_on_connect:
+            event.listen(pool, "connect", builtin_on_connect)
+
         def first_connect(dbapi_connection, connection_record):
             c = base.Connection(
                 engine,
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index d670cf231..9a138e69e 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -239,6 +239,7 @@ class DefaultDialect(interfaces.Dialect):
         self,
         encoding="utf-8",
         paramstyle=None,
+        isolation_level=None,
         dbapi=None,
         implicit_returning=None,
         supports_native_boolean=None,
@@ -251,12 +252,6 @@ class DefaultDialect(interfaces.Dialect):
         **kwargs
     ):
 
-        if not getattr(self, "ported_sqla_06", True):
-            util.warn(
-                "The %s dialect is not yet ported to the 0.6 format"
-                % self.name
-            )
-
         if server_side_cursors:
             if not self.supports_server_side_cursors:
                 raise exc.ArgumentError(
@@ -279,6 +274,7 @@ class DefaultDialect(interfaces.Dialect):
             self.implicit_returning = implicit_returning
         self.positional = self.paramstyle in ("qmark", "format", "numeric")
         self.identifier_preparer = self.preparer(self)
+        self._on_connect_isolation_level = isolation_level
         self.type_compiler = self.type_compiler(self)
         if supports_native_boolean is not None:
             self.supports_native_boolean = supports_native_boolean
@@ -345,6 +341,18 @@ class DefaultDialect(interfaces.Dialect):
         except ImportError:
             pass
 
+    def _builtin_onconnect(self):
+        if self._on_connect_isolation_level is not None:
+
+            def builtin_connect(dbapi_conn, conn_rec):
+                self._assert_and_set_isolation_level(
+                    dbapi_conn, self._on_connect_isolation_level
+                )
+
+            return builtin_connect
+        else:
+            return None
+
     def initialize(self, connection):
         try:
             self.server_version_info = self._get_server_version_info(
@@ -573,11 +581,51 @@ class DefaultDialect(interfaces.Dialect):
     def is_disconnect(self, e, connection, cursor):
         return False
 
+    @util.memoized_instancemethod
+    def _gen_allowed_isolation_levels(self, dbapi_conn):
+
+        try:
+            raw_levels = list(self.get_isolation_level_values(dbapi_conn))
+        except NotImplementedError:
+            return None
+        else:
+            normalized_levels = [
+                level.replace("_", " ").upper() for level in raw_levels
+            ]
+            if raw_levels != normalized_levels:
+                raise ValueError(
+                    f"Dialect {self.name!r} get_isolation_level_values() "
+                    f"method should return names as UPPERCASE using spaces, "
+                    f"not underscores; got "
+                    f"{sorted(set(raw_levels).difference(normalized_levels))}"
+                )
+            return tuple(normalized_levels)
+
+    def _assert_and_set_isolation_level(self, dbapi_conn, level):
+        level = level.replace("_", " ").upper()
+
+        _allowed_isolation_levels = self._gen_allowed_isolation_levels(
+            dbapi_conn
+        )
+        if (
+            _allowed_isolation_levels
+            and level not in _allowed_isolation_levels
+        ):
+            raise exc.ArgumentError(
+                f"Invalid value {level!r} for isolation_level. "
+                f"Valid isolation levels for {self.name!r} are "
+                f"{', '.join(_allowed_isolation_levels)}"
+            )
+
+        self.set_isolation_level(dbapi_conn, level)
+
     def reset_isolation_level(self, dbapi_conn):
         # default_isolation_level is read from the first connection
         # after the initial set of 'isolation_level', if any, so is
         # the configured default of this dialect.
-        self.set_isolation_level(dbapi_conn, self.default_isolation_level)
+        self._assert_and_set_isolation_level(
+            dbapi_conn, self.default_isolation_level
+        )
 
     def normalize_name(self, name):
         if name is None:
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 38d2c7a57..fbdf9b829 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -919,6 +919,11 @@ class Dialect(object):
         isolation level facilities; these APIs should be preferred for
         most typical use cases.
 
+        If the dialect also implements the
+        :meth:`.Dialect.get_isolation_level_values` method, then the given
+        level is guaranteed to be one of the string names within that sequence,
+        and the method will not need to anticipate a lookup failure.
+
         .. seealso::
 
             :meth:`_engine.Connection.get_isolation_level`
@@ -990,6 +995,48 @@ class Dialect(object):
         """
         raise NotImplementedError()
 
+    def get_isolation_level_values(self, dbapi_conn):
+        """return a sequence of string isolation level names that are accepted
+        by this dialect.
+
+        The available names should use the following conventions:
+
+        * use UPPERCASE names.   isolation level methods will accept lowercase
+          names but these are normalized into UPPERCASE before being passed
+          along to the dialect.
+        * separate words should be separated by spaces, not underscores, e.g.
+          ``REPEATABLE READ``.  isolation level names will have underscores
+          converted to spaces before being passed along to the dialect.
+        * The names for the four standard isolation names to the extent that
+          they are supported by the backend should be ``READ UNCOMMITTED``
+          ``READ COMMITTED``, ``REPEATABLE READ``, ``SERIALIZABLE``
+        * if the dialect supports an autocommit option it should be provided
+          using the isolation level name ``AUTOCOMMIT``.
+        * Other isolation modes may also be present, provided that they
+          are named in UPPERCASE and use spaces not underscores.
+
+        This function is used so that the default dialect can check that
+        a given isolation level parameter is valid, else raises an
+        :class:`_exc.ArgumentError`.
+
+        A DBAPI connection is passed to the method, in the unlikely event that
+        the dialect needs to interrogate the connection itself to determine
+        this list, however it is expected that most backends will return
+        a hardcoded list of values.  If the dialect supports "AUTOCOMMIT",
+        that value should also be present in the sequence returned.
+
+        The method raises ``NotImplementedError`` by default.  If a dialect
+        does not implement this method, then the default dialect will not
+        perform any checking on a given isolation level value before passing
+        it onto the :meth:`.Dialect.set_isolation_level` method.  This is
+        to allow backwards-compatibility with third party dialects that may
+        not yet be implementing this method.
+
+        .. versionadded:: 2.0
+
+        """
+        raise NotImplementedError()
+
     @classmethod
     def get_dialect_cls(cls, url):
         """Given a URL, return the :class:`.Dialect` that will be used.
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index c30fdf823..3aa2649f4 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -384,9 +384,13 @@ def _expect_raises(except_cls, msg=None, check_context=False):
         ec.error = err
         success = True
         if msg is not None:
-            assert re.search(
-                msg, util.text_type(err), re.UNICODE
-            ), "%r !~ %s" % (msg, err)
+            # I'm often pdbing here, and "err" above isn't
+            # in scope, so assign the string explicitly
+            error_as_string = util.text_type(err)
+            assert re.search(msg, error_as_string, re.UNICODE), "%r !~ %s" % (
+                msg,
+                error_as_string,
+            )
         if check_context and not are_we_already_in_a_traceback:
             _assert_proper_exception_context(err)
         print(util.text_type(err).encode("utf-8"))
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 4cc431bb7..a3d277c50 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -20,6 +20,7 @@ import sys
 
 from . import exclusions
 from . import only_on
+from .. import create_engine
 from .. import util
 from ..pool import QueuePool
 
@@ -873,6 +874,59 @@ class SuiteRequirements(Requirements):
                 ]
             }
         """
+        with config.db.connect() as conn:
+
+            try:
+                supported = conn.dialect.get_isolation_level_values(
+                    conn.connection.dbapi_connection
+                )
+            except NotImplementedError:
+                return None
+            else:
+                return {
+                    "default": conn.dialect.default_isolation_level,
+                    "supported": supported,
+                }
+
+    @property
+    def get_isolation_level_values(self):
+        """target dialect supports the
+        :meth:`_engine.Dialect.get_isolation_level_values`
+        method added in SQLAlchemy 2.0.
+
+        """
+
+        def go(config):
+            with config.db.connect() as conn:
+                try:
+                    conn.dialect.get_isolation_level_values(
+                        conn.connection.dbapi_connection
+                    )
+                except NotImplementedError:
+                    return False
+                else:
+                    return True
+
+        return exclusions.only_if(go)
+
+    @property
+    def dialect_level_isolation_level_param(self):
+        """test that the dialect allows the 'isolation_level' argument
+        to be handled by DefaultDialect"""
+
+        def go(config):
+            try:
+                e = create_engine(
+                    config.db.url, isolation_level="READ COMMITTED"
+                )
+            except:
+                return False
+            else:
+                return (
+                    e.dialect._on_connect_isolation_level == "READ COMMITTED"
+                )
+
+        return exclusions.only_if(go)
 
     @property
     def json_type(self):
diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index b3e43aad0..28fd99876 100644
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -8,6 +8,7 @@ from .. import eq_
 from .. import fixtures
 from .. import ne_
 from .. import provide_metadata
+from ..assertions import expect_raises_message
 from ..config import requirements
 from ..provision import set_default_schema_on_connection
 from ..schema import Column
@@ -140,6 +141,48 @@ class IsolationLevelTest(fixtures.TestBase):
                     levels["default"],
                 )
 
+    @testing.requires.get_isolation_level_values
+    def test_invalid_level_execution_option(self, connection_no_trans):
+        """test for the new get_isolation_level_values() method"""
+
+        connection = connection_no_trans
+        with expect_raises_message(
+            exc.ArgumentError,
+            "Invalid value '%s' for isolation_level. "
+            "Valid isolation levels for '%s' are %s"
+            % (
+                "FOO",
+                connection.dialect.name,
+                ", ".join(
+                    requirements.get_isolation_levels(config)["supported"]
+                ),
+            ),
+        ):
+            connection.execution_options(isolation_level="FOO")
+
+    @testing.requires.get_isolation_level_values
+    @testing.requires.dialect_level_isolation_level_param
+    def test_invalid_level_engine_param(self, testing_engine):
+        """test for the new get_isolation_level_values() method
+        and support for the dialect-level 'isolation_level' parameter.
+
+        """
+
+        eng = testing_engine(options=dict(isolation_level="FOO"))
+        with expect_raises_message(
+            exc.ArgumentError,
+            "Invalid value '%s' for isolation_level. "
+            "Valid isolation levels for '%s' are %s"
+            % (
+                "FOO",
+                eng.dialect.name,
+                ", ".join(
+                    requirements.get_isolation_levels(config)["supported"]
+                ),
+            ),
+        ):
+            eng.connect()
+
 
 class AutocommitIsolationTest(fixtures.TablesTest):
 
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index f1efaf7a6..ca203ad8c 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -54,7 +54,6 @@ from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import mock
-from sqlalchemy.testing.assertions import expect_raises_message
 from sqlalchemy.types import Boolean
 from sqlalchemy.types import Date
 from sqlalchemy.types import DateTime
@@ -597,21 +596,6 @@ class DialectTest(
                 )
             )
 
-    @testing.only_on("sqlite+pysqlite")
-    def test_isolation_level_message(self):
-        # needs to test that all three words are present and we also
-        # dont want to default all isolation level messages to use
-        # sorted(), so rely on python 3.7 for ordering of keywords
-        # in the message
-        with expect_raises_message(
-            exc.ArgumentError,
-            "Invalid value 'invalid' for "
-            "isolation_level. Valid isolation levels for "
-            "sqlite are READ UNCOMMITTED, SERIALIZABLE, AUTOCOMMIT",
-        ):
-            with testing.db.connect() as conn:
-                conn.execution_options(isolation_level="invalid")
-
     @testing.only_on("sqlite+pysqlcipher")
     def test_pysqlcipher_connects(self):
         """test #6586"""
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 50da425bd..e18f1c446 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -1102,6 +1102,8 @@ class AutoRollbackTest(fixtures.TestBase):
 
 
 class IsolationLevelTest(fixtures.TestBase):
+    """see also sqlalchemy/testing/suite/test_dialect.py::IsolationLevelTest"""
+
     __requires__ = (
         "isolation_level",
         "ad_hoc_engines",
@@ -1123,7 +1125,6 @@ class IsolationLevelTest(fixtures.TestBase):
         else:
             assert False, "no non-default isolation level available"
 
-    @testing.requires.legacy_isolation_level
     def test_engine_param_stays(self):
 
         eng = testing_engine()
@@ -1177,7 +1178,6 @@ class IsolationLevelTest(fixtures.TestBase):
 
         conn.close()
 
-    @testing.requires.legacy_isolation_level
     def test_reset_level_with_setting(self):
         eng = testing_engine(
             options=dict(isolation_level=self._non_default_isolation_level())
@@ -1201,45 +1201,88 @@ class IsolationLevelTest(fixtures.TestBase):
         )
         conn.close()
 
-    @testing.requires.legacy_isolation_level
+    def test_underscore_replacement(self, connection_no_trans):
+        conn = connection_no_trans
+        with mock.patch.object(
+            conn.dialect, "set_isolation_level"
+        ) as mock_sil, mock.patch.object(
+            conn.dialect,
+            "_gen_allowed_isolation_levels",
+            mock.Mock(return_value=["READ COMMITTED", "REPEATABLE READ"]),
+        ):
+            conn.execution_options(isolation_level="REPEATABLE_READ")
+            dbapi_conn = conn.connection.dbapi_connection
+
+        eq_(mock_sil.mock_calls, [mock.call(dbapi_conn, "REPEATABLE READ")])
+
+    def test_casing_replacement(self, connection_no_trans):
+        conn = connection_no_trans
+        with mock.patch.object(
+            conn.dialect, "set_isolation_level"
+        ) as mock_sil, mock.patch.object(
+            conn.dialect,
+            "_gen_allowed_isolation_levels",
+            mock.Mock(return_value=["READ COMMITTED", "REPEATABLE READ"]),
+        ):
+            conn.execution_options(isolation_level="repeatable_read")
+            dbapi_conn = conn.connection.dbapi_connection
+
+        eq_(mock_sil.mock_calls, [mock.call(dbapi_conn, "REPEATABLE READ")])
+
+    def test_dialect_doesnt_follow_naming_guidelines(
+        self, connection_no_trans
+    ):
+        conn = connection_no_trans
+
+        conn.dialect.__dict__.pop("_gen_allowed_isolation_levels", None)
+        with mock.patch.object(
+            conn.dialect,
+            "get_isolation_level_values",
+            mock.Mock(
+                return_value=[
+                    "READ COMMITTED",
+                    "REPEATABLE_READ",
+                    "serializable",
+                ]
+            ),
+        ):
+            with expect_raises_message(
+                ValueError,
+                f"Dialect {conn.dialect.name!r} "
+                r"get_isolation_level_values\(\) method "
+                r"should "
+                r"return names as UPPERCASE using spaces, not underscores; "
+                r"got \['REPEATABLE_READ', 'serializable'\]",
+            ):
+                conn.execution_options(isolation_level="READ COMMITTED")
+
     def test_invalid_level_engine_param(self):
         eng = testing_engine(options=dict(isolation_level="FOO"))
         assert_raises_message(
             exc.ArgumentError,
-            "Invalid value '%s' for isolation_level. "
-            "Valid isolation levels for %s are %s"
-            % (
-                "FOO",
-                eng.dialect.name,
-                ", ".join(eng.dialect._isolation_lookup),
-            ),
+            f"Invalid value 'FOO' for isolation_level. "
+            f"Valid isolation levels for {eng.dialect.name!r} are "
+            f"""{', '.join(
+                testing.requires.get_isolation_levels(
+                    testing.config
+                )['supported']
+            )}""",
             eng.connect,
         )
 
-    # TODO: all the dialects seem to be manually raising ArgumentError
-    # individually within their set_isolation_level() methods, when this
-    # should be a default dialect feature so that
-    # error messaging etc. is consistent, including that it works for 3rd
-    # party dialects.
-    # TODO: barring that, at least implement this for the Oracle dialect
-    @testing.fails_on(
-        "oracle",
-        "cx_oracle dialect doesnt have argument error here, "
-        "raises it via the DB rejecting it",
-    )
     def test_invalid_level_execution_option(self):
         eng = testing_engine(
             options=dict(execution_options=dict(isolation_level="FOO"))
         )
         assert_raises_message(
             exc.ArgumentError,
-            "Invalid value '%s' for isolation_level. "
-            "Valid isolation levels for %s are %s"
-            % (
-                "FOO",
-                eng.dialect.name,
-                ", ".join(eng.dialect._isolation_lookup),
-            ),
+            f"Invalid value 'FOO' for isolation_level. "
+            f"Valid isolation levels for {eng.dialect.name!r} are "
+            f"""{', '.join(
+                testing.requires.get_isolation_levels(
+                    testing.config
+                )['supported']
+            )}""",
             eng.connect,
         )
 
diff --git a/test/requirements.py b/test/requirements.py
index 28a283b12..378d54fbd 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -383,41 +383,6 @@ class DefaultRequirements(SuiteRequirements):
         )
 
     @property
-    def legacy_isolation_level(self):
-        # refers dialects where "isolation_level" can be passed to
-        # create_engine
-        return only_on(
-            ("postgresql", "sqlite", "mysql", "mariadb", "mssql"),
-            "DBAPI has no isolation level support",
-        )
-
-    def get_isolation_levels(self, config):
-        levels = set(config.db.dialect._isolation_lookup)
-
-        if against(config, "sqlite"):
-            default = "SERIALIZABLE"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "postgresql"):
-            default = "READ COMMITTED"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "mysql"):
-            default = "REPEATABLE READ"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "mariadb"):
-            default = "REPEATABLE READ"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "mssql"):
-            default = "READ COMMITTED"
-            levels.add("AUTOCOMMIT")
-        elif against(config, "oracle"):
-            default = "READ COMMITTED"
-            levels.add("AUTOCOMMIT")
-        else:
-            raise NotImplementedError()
-
-        return {"default": default, "supported": levels}
-
-    @property
     def autocommit(self):
         """target dialect supports 'AUTOCOMMIT' as an isolation_level"""