Narrow exception handling in do_rollback and do_commit

Catch specific OperationalError types instead of bare Exception to
prevent swallowing unrelated errors. Also use case-insensitive string
matching and handle both dqlitedbapi and dqliteclient exception types.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -72,11 +72,16 @@ def do_rollback(self, dbapi_connection: DBAPIConnection) -> None:
         dqlite throws an error if we try to rollback when no transaction
         is active, so we catch and ignore that specific error.
         """
+        import dqliteclient.exceptions
+        import dqlitedbapi.exceptions
+
         try:
             dbapi_connection.rollback()
-        except Exception as e:
-            # Ignore "no transaction is active" errors
-            if "no transaction is active" not in str(e):
+        except (
+            dqlitedbapi.exceptions.OperationalError,
+            dqliteclient.exceptions.OperationalError,
+        ) as e:
+            if "no transaction is active" not in str(e).lower():
                 raise
 
     def do_commit(self, dbapi_connection: DBAPIConnection) -> None:
@@ -85,11 +90,16 @@ def do_commit(self, dbapi_connection: DBAPIConnection) -> None:
         dqlite throws an error if we try to commit when no transaction
         is active, so we catch and ignore that specific error.
         """
+        import dqliteclient.exceptions
+        import dqlitedbapi.exceptions
+
         try:
             dbapi_connection.commit()
-        except Exception as e:
-            # Ignore "no transaction is active" errors
-            if "no transaction is active" not in str(e):
+        except (
+            dqlitedbapi.exceptions.OperationalError,
+            dqliteclient.exceptions.OperationalError,
+        ) as e:
+            if "no transaction is active" not in str(e).lower():
                 raise
 
     _dqlite_disconnect_messages = (

tests/test_dialect.py

@@ -152,6 +152,86 @@ def test_async_dialect_returns_underlying_connection(self) -> None:
         )
 
 
+class TestDoRollbackCommit:
+    def test_do_rollback_catches_only_operational_error(self) -> None:
+        """do_rollback should only catch OperationalError, not bare Exception."""
+        import ast
+        import inspect
+        import textwrap
+
+        source = textwrap.dedent(inspect.getsource(DqliteDialect.do_rollback))
+        tree = ast.parse(source)
+
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ExceptHandler):
+                assert node.type is not None, "do_rollback uses bare except"
+                if isinstance(node.type, ast.Name):
+                    assert node.type.id != "Exception", (
+                        "do_rollback catches bare Exception; "
+                        "should catch a specific dqlite exception type"
+                    )
+
+    def test_do_commit_catches_only_operational_error(self) -> None:
+        """do_commit should only catch OperationalError, not bare Exception."""
+        import ast
+        import inspect
+        import textwrap
+
+        source = textwrap.dedent(inspect.getsource(DqliteDialect.do_commit))
+        tree = ast.parse(source)
+
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ExceptHandler):
+                assert node.type is not None, "do_commit uses bare except"
+                if isinstance(node.type, ast.Name):
+                    assert node.type.id != "Exception", (
+                        "do_commit catches bare Exception; "
+                        "should catch a specific dqlite exception type"
+                    )
+
+    def test_do_rollback_suppresses_no_transaction_error(self) -> None:
+        """do_rollback should suppress 'no transaction is active' errors."""
+        from unittest.mock import MagicMock
+
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        mock_conn = MagicMock()
+        mock_conn.rollback.side_effect = dqlitedbapi.exceptions.OperationalError(
+            "no transaction is active"
+        )
+        dialect.do_rollback(mock_conn)
+
+    def test_do_rollback_suppresses_client_no_transaction_error(self) -> None:
+        """do_rollback should suppress dqliteclient 'no transaction' errors too."""
+        from unittest.mock import MagicMock
+
+        import dqliteclient.exceptions
+
+        dialect = DqliteDialect()
+        mock_conn = MagicMock()
+        mock_conn.rollback.side_effect = dqliteclient.exceptions.OperationalError(
+            1, "cannot rollback - no transaction is active"
+        )
+        dialect.do_rollback(mock_conn)
+
+    def test_do_rollback_raises_other_errors(self) -> None:
+        """do_rollback should re-raise errors other than 'no transaction'."""
+        from unittest.mock import MagicMock
+
+        import pytest
+
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        mock_conn = MagicMock()
+        mock_conn.rollback.side_effect = dqlitedbapi.exceptions.OperationalError(
+            "database is locked"
+        )
+        with pytest.raises(dqlitedbapi.exceptions.OperationalError, match="database is locked"):
+            dialect.do_rollback(mock_conn)
+
+
 class TestIsDisconnect:
     def test_recognizes_connection_closed(self) -> None:
         """is_disconnect should return True for connection-closed errors."""