Fix Azure Synapse connection error

Fixed regression caused by the fix for :ticket:`8231` released in 1.4.40
where connection would fail if the user does not have permission to query
the dm_exec_sessions or dm_pdw_nodes_exec_sessions system view when trying
to determine the current transaction isolation level.

Fixes: #8475
Change-Id: Ie2bcda92f2ef2d12360ddda47eb6e896313c71f2

3 files changed, 52 insertions, 19 deletions

diff --git a/doc/build/changelog/unreleased_14/8475.rst b/doc/build/changelog/unreleased_14/8475.rst
new file mode 100644
index 000000000..22fc3f2dd
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8475.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, mssql, regression
+    :tickets: 8475
+
+    Fixed regression caused by the fix for :ticket:`8231` released in 1.4.40
+    where connection would fail if the user does not have permission to query
+    the dm_exec_sessions or dm_pdw_nodes_exec_sessions system view when trying
+    to determine the current transaction isolation level.
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 94e082611..82a5bb6f7 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -3008,27 +3008,34 @@ class MSDialect(default.DefaultDialect):
                 )
 
             view_name = "sys.{}".format(row[0])
-            cursor.execute(
-                """
-                    SELECT CASE transaction_isolation_level
-                    WHEN 0 THEN NULL
-                    WHEN 1 THEN 'READ UNCOMMITTED'
-                    WHEN 2 THEN 'READ COMMITTED'
-                    WHEN 3 THEN 'REPEATABLE READ'
-                    WHEN 4 THEN 'SERIALIZABLE'
-                    WHEN 5 THEN 'SNAPSHOT' END AS TRANSACTION_ISOLATION_LEVEL
-                    FROM {}
-                    where session_id = @@SPID
-                """.format(
-                    view_name
+
+            try:
+                cursor.execute(
+                    """
+                        SELECT CASE transaction_isolation_level
+                        WHEN 0 THEN NULL
+                        WHEN 1 THEN 'READ UNCOMMITTED'
+                        WHEN 2 THEN 'READ COMMITTED'
+                        WHEN 3 THEN 'REPEATABLE READ'
+                        WHEN 4 THEN 'SERIALIZABLE'
+                        WHEN 5 THEN 'SNAPSHOT' END
+                        AS TRANSACTION_ISOLATION_LEVEL
+                        FROM {}
+                        where session_id = @@SPID
+                    """.format(
+                        view_name
+                    )
                 )
-            )
-            row = cursor.fetchone()
-            assert row is not None
-            val = row[0]
+            except self.dbapi.Error as err:
+                raise NotImplementedError(
+                    "Can't fetch isolation level;  encountered error {} when "
+                    'attempting to query the "{}" view.'.format(err, view_name)
+                ) from err
+            else:
+                row = cursor.fetchone()
+                return row[0].upper()
         finally:
             cursor.close()
-        return val.upper()
 
     def initialize(self, connection):
         super(MSDialect, self).initialize(connection)
diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py
index 028920203..28c0fb532 100644
--- a/test/dialect/mssql/test_engine.py
+++ b/test/dialect/mssql/test_engine.py
@@ -647,7 +647,7 @@ class RealIsolationLevelTest(fixtures.TestBase):
 
 
 class IsolationLevelDetectTest(fixtures.TestBase):
-    def _fixture(self, view_result):
+    def _fixture(self, view_result, simulate_perm_failure=False):
         class Error(Exception):
             pass
 
@@ -670,6 +670,10 @@ class IsolationLevelDetectTest(fixtures.TestBase):
                 stmt,
                 re.S,
             ):
+                if simulate_perm_failure:
+                    raise dialect.dbapi.Error(
+                        "SQL Server simulated permission error"
+                    )
                 result.append(("SERIALIZABLE",))
             else:
                 assert False
@@ -705,6 +709,20 @@ class IsolationLevelDetectTest(fixtures.TestBase):
             connection,
         )
 
+    def test_dont_have_table_perms(self):
+        dialect, connection = self._fixture(
+            "dm_pdw_nodes_exec_sessions", simulate_perm_failure=True
+        )
+
+        assert_raises_message(
+            NotImplementedError,
+            r"Can\'t fetch isolation level;  encountered error SQL Server "
+            r"simulated permission error when attempting to query the "
+            r'"sys.dm_pdw_nodes_exec_sessions" view.',
+            dialect.get_isolation_level,
+            connection,
+        )
+
 
 class InvalidTransactionFalsePositiveTest(fixtures.TablesTest):
     __only_on__ = "mssql"