Implement autocommit isolation level for cx_oracle

As part of this change Oracle also gets the concept of a
default isolation level, however since Oracle does not provide a
fixed method of knowing what the isolation level would be without a
server side transaction actually in progress, for now we hardcode
just to "READ COMMITTED".

Enhanced the test suite for isolation level testing in the dialect
test suite and added features to requirements so that the supported
isolation levels can be reported generically for dialects.

Co-authored-by: Mike Bayer <mike_mp@zzzcomputing.com>
Fixes: #5200
Change-Id: I2c4d49da9ff80ccc228c21e196ec9a961de53478

4 files changed, 149 insertions, 1 deletions

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index d12453924..c0ee66ad7 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -53,6 +53,36 @@ This step is also required when using table reflection, i.e. autoload=True::
         autoload=True
   )
 
+Transaction Isolation Level / Autocommit
+----------------------------------------
+
+The Oracle database supports "READ COMMITTED" and "SERIALIZABLE" modes
+of isolation, however the SQLAlchemy Oracle dialect currently only has
+explicit support for "READ COMMITTED".  It is possible to emit a
+"SET TRANSACTION" statement on a connection in order to use SERIALIZABLE
+isolation, however the SQLAlchemy dialect will remain unaware of this setting,
+such as if the :meth:`.Connection.get_isolation_level` method is used;
+this method is hardcoded to return "READ COMMITTED" right now.
+
+The AUTOCOMMIT isolation level is also supported by the cx_Oracle dialect.
+
+To set using per-connection execution options::
+
+    connection = engine.connect()
+    connection = connection.execution_options(
+        isolation_level="AUTOCOMMIT"
+    )
+
+Valid values for ``isolation_level`` include:
+
+* ``READ COMMITTED``
+* ``AUTOCOMMIT``
+
+
+.. versionadded:: 1.3.16 added support for AUTOCOMMIT to the cx_oracle dialect
+   as well as the notion of a default isolation level, currently harcoded
+   to "READ COMMITTED".
+
 Identifier Casing
 -----------------
 
@@ -1395,6 +1425,20 @@ class OracleDialect(default.DefaultDialect):
             connection, additional_tests
         )
 
+    _isolation_lookup = ["READ COMMITTED"]
+
+    def get_isolation_level(self, connection):
+        return "READ COMMITTED"
+
+    def set_isolation_level(self, connection, level):
+        # prior to adding AUTOCOMMIT support for cx_Oracle, the Oracle dialect
+        # had no notion of setting the isolation level.  As Oracle
+        # does not have a straightforward way of getting the isolation level
+        # if a server-side transaction is not yet in progress, we currently
+        # hardcode to only support "READ COMMITTED" and "AUTOCOMMIT" at the
+        # cx_oracle level.  See #5200.
+        pass
+
     def has_table(self, connection, table_name, schema=None):
         if not schema:
             schema = self.default_schema_name
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index d5177daa6..3a3bbad25 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -1138,5 +1138,18 @@ class OracleDialect_cx_oracle(OracleDialect):
     def do_recover_twophase(self, connection):
         connection.info.pop("cx_oracle_prepared", None)
 
+    def set_isolation_level(self, connection, level):
+        if hasattr(connection, "connection"):
+            dbapi_connection = connection.connection
+        else:
+            dbapi_connection = connection
+        if level == "AUTOCOMMIT":
+            dbapi_connection.autocommit = True
+        else:
+            dbapi_connection.autocommit = False
+            super(OracleDialect_cx_oracle, self).set_isolation_level(
+                dbapi_connection, level
+            )
+
 
 dialect = OracleDialect_cx_oracle
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 5eb769bf3..5ee0d67a2 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -703,6 +703,42 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def isolation_level(self):
+        """target dialect supports general isolation level settings.
+
+        Note that this requirement, when enabled, also requires that
+        the get_isolation_levels() method be implemented.
+
+        """
+        return exclusions.closed()
+
+    def get_isolation_levels(self, config):
+        """Return a structure of supported isolation levels for the current
+        testing dialect.
+
+        The structure indicates to the testing suite what the expected
+        "default" isolation should be, as well as the other values that
+        are accepted.  The dictionary has two keys, "default" and "supported".
+        The "supported" key refers to a list of all supported levels and
+        it should include AUTOCOMMIT if the dialect supports it.
+
+        If the :meth:`.DefaultRequirements.isolation_level` requirement is
+        not open, then this method has no return value.
+
+        E.g.::
+
+            >>> testing.requirements.get_isolation_levels()
+            {
+                "default": "READ_COMMITED",
+                "supported": [
+                    "SERIALIZABLE", "READ UNCOMMITTED",
+                    "READ COMMITTED", "REPEATABLE READ",
+                    "AUTOCOMMIT"
+                ]
+            }
+        """
+
+    @property
     def json_type(self):
         """target platform implements a native JSON type."""
 
diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py
index 1b3307042..8aa13a622 100644
--- a/lib/sqlalchemy/testing/suite/test_dialect.py
+++ b/lib/sqlalchemy/testing/suite/test_dialect.py
@@ -4,6 +4,7 @@ from .. import assert_raises
 from .. import config
 from .. import eq_
 from .. import fixtures
+from .. import ne_
 from .. import provide_metadata
 from ..config import requirements
 from ..schema import Column
@@ -79,6 +80,46 @@ class ExceptionTest(fixtures.TablesTest):
                 assert isinstance(err_str, str)
 
 
+class IsolationLevelTest(fixtures.TestBase):
+    __backend__ = True
+
+    __requires__ = ("isolation_level",)
+
+    def _get_non_default_isolation_level(self):
+        levels = requirements.get_isolation_levels(config)
+
+        default = levels["default"]
+        supported = levels["supported"]
+
+        s = set(supported).difference(["AUTOCOMMIT", default])
+        if s:
+            return s.pop()
+        else:
+            config.skip_test("no non-default isolation level available")
+
+    def test_default_isolation_level(self):
+        eq_(
+            config.db.dialect.default_isolation_level,
+            requirements.get_isolation_levels(config)["default"],
+        )
+
+    def test_non_default_isolation_level(self):
+        non_default = self._get_non_default_isolation_level()
+
+        with config.db.connect() as conn:
+            existing = conn.get_isolation_level()
+
+            ne_(existing, non_default)
+
+            conn.execution_options(isolation_level=non_default)
+
+            eq_(conn.get_isolation_level(), non_default)
+
+            conn.dialect.reset_isolation_level(conn.connection)
+
+            eq_(conn.get_isolation_level(), existing)
+
+
 class AutocommitTest(fixtures.TablesTest):
 
     run_deletes = "each"
@@ -115,13 +156,27 @@ class AutocommitTest(fixtures.TablesTest):
         conn = config.db.connect()
         c2 = conn.execution_options(isolation_level="AUTOCOMMIT")
         self._test_conn_autocommits(c2, True)
-        conn.invalidate()
+
+        c2.dialect.reset_isolation_level(c2.connection)
+
         self._test_conn_autocommits(conn, False)
 
     def test_autocommit_off(self):
         conn = config.db.connect()
         self._test_conn_autocommits(conn, False)
 
+    def test_turn_autocommit_off_via_default_iso_level(self):
+        conn = config.db.connect()
+        conn.execution_options(isolation_level="AUTOCOMMIT")
+        self._test_conn_autocommits(conn, True)
+
+        conn.execution_options(
+            isolation_level=requirements.get_isolation_levels(config)[
+                "default"
+            ]
+        )
+        self._test_conn_autocommits(conn, False)
+
 
 class EscapingTest(fixtures.TestBase):
     @provide_metadata