Fix failed db.execute() write leaves a phantom transaction open

Refs https://github.com/simonw/sqlite-utils/issues/769#issuecomment-4900034150

docs/changelog.rst

@@ -16,6 +16,7 @@ Unreleased
 - ``--no-headers`` now omits the header row from ``--fmt`` and ``--table`` output, not just CSV and TSV output. (:issue:`566`)
 - ``table.insert_all(..., pk=...)`` now raises ``InvalidColumns`` if ``pk=`` names columns that do not exist in an existing table. Previously this behaved inconsistently, with single-row inserts raising a ``KeyError`` while other row counts succeeded. (:issue:`732`)
 - Fixed an ``IndexError`` from ``table.insert(..., pk=..., ignore=True)`` when an ignored insert followed writes to another table on the same connection. ``last_pk`` is now populated from the explicit primary key value instead of looking up a stale ``lastrowid``. (:issue:`554`)
+- Fixed a bug where a failed write statement executed with ``db.execute()`` left the driver's implicit transaction open. Every subsequent write then joined that phantom transaction, which nothing committed, so their work was silently rolled back when the connection was closed. The implicit transaction opened by a failed statement is now rolled back before the exception is raised. A failed write inside a transaction opened with ``db.begin()`` or ``db.atomic()`` leaves that transaction open and untouched, as before.
 
 .. _v4_0rc3:
 

sqlite_utils/db.py

@@ -870,10 +870,18 @@ class Database:
         if self._tracer:
             self._tracer(sql, parameters)
         was_in_transaction = self.conn.in_transaction
-        if parameters is not None:
-            cursor = self.conn.execute(sql, parameters)
-        else:
-            cursor = self.conn.execute(sql)
+        try:
+            if parameters is not None:
+                cursor = self.conn.execute(sql, parameters)
+            else:
+                cursor = self.conn.execute(sql)
+        except Exception:
+            if not was_in_transaction and self.conn.in_transaction:
+                # The failed statement opened an implicit transaction that
+                # nothing would ever commit - roll it back, otherwise it
+                # would capture every subsequent write
+                self.conn.execute("ROLLBACK")
+            raise
         if (
             not was_in_transaction
             and self.conn.in_transaction

tests/test_atomic.py

@@ -258,6 +258,47 @@ def test_execute_comment_prefixed_begin_leaves_transaction_open(fresh_db):
     assert [r["id"] for r in fresh_db["t"].rows] == [1]
 
 
+def test_execute_failed_write_rolls_back_implicit_transaction(tmpdir):
+    # A failed write must not leave the driver's implicit transaction open -
+    # that would silently disable auto-commit for every subsequent write
+    path = str(tmpdir / "test.db")
+    db = Database(path)
+    db["t"].insert({"id": 1}, pk="id")
+    with pytest.raises(sqlite3.IntegrityError):
+        db.execute("insert into t (id) values (1)")
+    assert not db.conn.in_transaction
+    # Subsequent writes commit as normal and survive closing the connection
+    db["other"].insert({"id": 2})
+    db.close()
+    db2 = Database(path)
+    assert db2["other"].exists()
+    db2.close()
+
+
+def test_execute_failed_write_preserves_explicit_transaction(fresh_db):
+    # A failed write inside an explicit transaction must not roll back
+    # the caller's earlier work - only the caller decides that
+    fresh_db["t"].insert({"id": 1}, pk="id")
+    fresh_db.begin()
+    fresh_db.execute("insert into t (id) values (2)")
+    with pytest.raises(sqlite3.IntegrityError):
+        fresh_db.execute("insert into t (id) values (1)")
+    assert fresh_db.conn.in_transaction
+    fresh_db.commit()
+    assert [r["id"] for r in fresh_db["t"].rows] == [1, 2]
+
+
+def test_execute_failed_write_inside_atomic_preserves_block(fresh_db):
+    # A caught failure inside an atomic() block must leave the block's
+    # transaction open so its other work still commits
+    fresh_db["t"].insert({"id": 1}, pk="id")
+    with fresh_db.atomic():
+        fresh_db.execute("insert into t (id) values (2)")
+        with pytest.raises(sqlite3.IntegrityError):
+            fresh_db.execute("insert into t (id) values (1)")
+    assert [r["id"] for r in fresh_db["t"].rows] == [1, 2]
+
+
 def test_query_returning_commits_after_iteration(tmpdir):
     if sqlite3.sqlite_version_info < (3, 35, 0):
         import pytest as _pytest