Open write task transactions explicitly with BEGIN IMMEDIATE

Write tasks with transaction=True previously relied on the sqlite3
driver's implicit BEGIN, which only fires on the first raw
data-modifying statement. sqlite-utils 4.0 write methods found no open
transaction, committed their own work mid-task, and Datasette's
commit/rollback at task end was a no-op - so a failing write function
could leave partial writes permanently committed.

The write thread (and the non-threaded write path) now executes BEGIN
IMMEDIATE before invoking each transaction=True task, commits when it
returns and rolls back if it raises. sqlite-utils methods nest inside
that transaction as savepoints, restoring task-level atomicity for
every write path.

execute_write() now detects statements SQLite refuses to run inside a
transaction (VACUUM, ATTACH, DETACH, PRAGMA) and runs those in
autocommit mode, preserving previous behavior for e.g. trusted canned
queries that run VACUUM.

Refs #2831

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N76afGMhBRQk528VF1LTpR

datasette/database.py

@@ -5,6 +5,7 @@ import inspect
 import os
 from pathlib import Path
 import queue
+import re
 import sqlite_utils
 import sys
 import tempfile
@@ -42,6 +43,39 @@ class DatasetteClosedError(RuntimeError):
 
 _SHUTDOWN = object()
 
+# Statements that SQLite refuses to execute inside a transaction. These run
+# without the task transaction - a single statement needs no extra atomicity.
+_STATEMENTS_DISALLOWED_IN_TRANSACTION = {"vacuum", "attach", "detach", "pragma"}
+
+_FIRST_KEYWORD_RE = re.compile(
+    r"^(?:\s+|--[^\n]*(?:\n|$)|/\*.*?\*/)*([a-zA-Z]+)", re.DOTALL
+)
+
+
+def _can_execute_in_transaction(sql):
+    match = _FIRST_KEYWORD_RE.match(sql)
+    if match is None:
+        return True
+    return match.group(1).lower() not in _STATEMENTS_DISALLOWED_IN_TRANSACTION
+
+
+def _run_write_in_transaction(conn, fn):
+    # Open the transaction explicitly instead of relying on the sqlite3
+    # driver's implicit BEGIN, which only fires on the first data-modifying
+    # statement. With a transaction genuinely open, sqlite-utils write
+    # methods nest inside it as savepoints instead of committing their own
+    # transactions mid-task: https://github.com/simonw/datasette/issues/2831
+    conn.execute("BEGIN IMMEDIATE")
+    try:
+        result = fn(conn)
+    except Exception:
+        if conn.in_transaction:
+            conn.rollback()
+        raise
+    if conn.in_transaction:
+        conn.commit()
+    return result
+
 
 class Database:
     # For table counts stop at this many rows:
@@ -258,7 +292,12 @@ class Database:
             )
 
         with trace("sql", database=self.name, sql=sql.strip(), params=params):
-            results = await self.execute_write_fn(_inner, block=block, request=request)
+            results = await self.execute_write_fn(
+                _inner,
+                block=block,
+                transaction=_can_execute_in_transaction(sql),
+                request=request,
+            )
         return results
 
     async def execute_write_script(self, sql, block=True, request=None):
@@ -347,8 +386,7 @@ class Database:
                 self._write_connection = self.connect(write=True)
                 self.ds._prepare_connection(self._write_connection, self.name)
             if transaction:
-                with self._write_connection:
-                    result = fn(self._write_connection)
+                result = _run_write_in_transaction(self._write_connection, fn)
             else:
                 result = fn(self._write_connection)
         else:
@@ -476,8 +514,7 @@ class Database:
             else:
                 try:
                     if task.transaction:
-                        with conn:
-                            result = task.fn(conn)
+                        result = _run_write_in_transaction(conn, task.fn)
                     else:
                         result = task.fn(conn)
                 except Exception as e:

docs/changelog.rst

@@ -4,6 +4,14 @@
 Changelog
 =========
 
+.. _v_unreleased:
+
+Unreleased
+----------
+
+- Write functions run via ``await db.execute_write_fn()`` now execute inside an explicitly opened ``BEGIN IMMEDIATE`` transaction, committed when the function returns or rolled back if it raises. Previously the transaction was only opened implicitly by the first raw data-modifying statement, which meant writes made through sqlite-utils committed independently mid-task - a function that used sqlite-utils and then failed could leave those writes permanently committed. sqlite-utils write methods now nest inside the task transaction as savepoints, so a failing write function rolls back everything it did. Functions run with ``transaction=True`` should no longer manage transactions themselves - use ``transaction=False`` for manual transaction control. (:issue:`2831`)
+- ``await db.execute_write()`` detects statements that SQLite cannot execute inside a transaction - ``VACUUM``, ``ATTACH``, ``DETACH`` and ``PRAGMA`` - and runs them in autocommit mode instead. (:issue:`2831`)
+
 .. _v1_0_a36:
 
 1.0a36 (2026-07-07)

docs/internals.rst

@@ -2059,7 +2059,7 @@ If you need to retrieve every row returned by a statement, pass ``return_all=Tru
 
 If you pass ``block=False`` this behavior changes to "fire and forget" - queries will be added to the write queue and executed in a separate thread while your code can continue to do other things. The method will return a UUID representing the queued task.
 
-Each call to ``execute_write()`` will be executed inside a transaction.
+Each call to ``execute_write()`` will be executed inside a transaction, with the exception of statements that SQLite does not allow to run inside a transaction: ``VACUUM``, ``ATTACH``, ``DETACH`` and ``PRAGMA``. Those statements are executed in autocommit mode instead.
 
 .. _database_execute_write_script:
 
@@ -2151,7 +2151,11 @@ The value returned from ``await database.execute_write_fn(...)`` will be the ret
 
 If your function raises an exception that exception will be propagated up to the ``await`` line.
 
-By default your function will be executed inside a transaction. You can pass ``transaction=False`` to disable this behavior, though if you do that you should be careful to manually apply transactions - ideally using the ``with conn:`` pattern, or you may see ``OperationalError: database table is locked`` errors.
+By default your function will be executed inside a transaction. Datasette executes ``BEGIN IMMEDIATE`` on the write connection before calling your function, then commits the transaction when your function returns - or rolls it back if your function raises an exception. Nothing your function writes will be visible to other connections until that final commit.
+
+Because the transaction is already open when your function is called, write methods from libraries such as `sqlite-utils <https://sqlite-utils.datasette.io/>`__ will nest their work inside it (sqlite-utils uses savepoints) rather than committing independently, so an exception rolls back everything the function did.
+
+Your function should not manage transactions itself when ``transaction=True`` - do not execute ``BEGIN`` or call ``conn.commit()`` or ``conn.rollback()`` on the connection. If you need to manage transactions manually, pass ``transaction=False`` - ideally using the ``with conn:`` pattern, or you may see ``OperationalError: database table is locked`` errors.
 
 If you specify ``block=False`` the method becomes fire-and-forget, queueing your function to be executed and then allowing your code after the call to ``.execute_write_fn()`` to continue running while the underlying thread waits for an opportunity to run your function. A UUID representing the queued task will be returned. Any exceptions in your code will be silently swallowed.
 

tests/test_internals_database.py

@@ -11,6 +11,7 @@ from datasette.database import _deliver_write_result
 from datasette.utils.sqlite import sqlite3, supports_returning
 from datasette.utils import Column
 import pytest
+import sqlite_utils
 import time
 import uuid
 
@@ -111,6 +112,99 @@ async def test_execute_fn_transaction_false():
     await db.execute_write_fn(run, transaction=False)
 
 
+@pytest.mark.asyncio
+async def test_execute_write_fn_wraps_sqlite_utils_writes_in_transaction():
+    # https://github.com/simonw/datasette/issues/2831
+    # sqlite-utils write methods commit their own transactions unless one is
+    # already open - the write thread must open one before running each task
+    # so that a failing task rolls back everything, including those writes.
+    datasette = Datasette(memory=True)
+    db = datasette.add_memory_database("test_txn_sqlite_utils")
+    await db.execute_write("create table t (a integer)")
+
+    def failing_task(conn):
+        sqlite_utils.Database(conn)["t"].insert_all({"a": i} for i in range(5))
+        assert conn.in_transaction
+        raise ValueError("boom")
+
+    with pytest.raises(ValueError):
+        await db.execute_write_fn(failing_task)
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 0
+    # The write connection should be back in autocommit mode
+    assert (
+        await db.execute_write_fn(lambda conn: conn.in_transaction, transaction=False)
+    ) is False
+    # And a transaction=True task should see a transaction already open
+    assert (await db.execute_write_fn(lambda conn: conn.in_transaction)) is True
+
+
+@pytest.mark.asyncio
+async def test_execute_write_statements_disallowed_in_transaction(tmp_path):
+    # VACUUM (and ATTACH/DETACH/PRAGMA) cannot run inside a transaction, so
+    # execute_write() must run them outside one
+    # https://github.com/simonw/datasette/issues/2831
+    path = str(tmp_path / "test.db")
+    setup_conn = sqlite3.connect(path)
+    setup_conn.execute("create table t (a integer)")
+    setup_conn.close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+    await db.execute_write("vacuum")
+    await db.execute_write("  -- a comment\n  VACUUM")
+    # But regular DML statements still run inside a transaction
+    from datasette.database import _can_execute_in_transaction
+
+    assert _can_execute_in_transaction("insert into t values (1)")
+    assert _can_execute_in_transaction("with x as (select 1) insert into t select 1")
+    assert not _can_execute_in_transaction("vacuum")
+    assert not _can_execute_in_transaction("/* hi */ PRAGMA optimize")
+
+
+@pytest.mark.asyncio
+async def test_execute_write_fn_sqlite_utils_integrity_error_rolls_back_task():
+    # https://github.com/simonw/datasette/issues/2831
+    datasette = Datasette(memory=True)
+    db = datasette.add_memory_database("test_txn_integrity")
+    await db.execute_write("create table t (id integer primary key)")
+
+    def two_inserts(conn):
+        table = sqlite_utils.Database(conn)["t"]
+        table.insert({"id": 1})
+        table.insert({"id": 1})  # IntegrityError
+
+    with pytest.raises(sqlite3.IntegrityError):
+        await db.execute_write_fn(two_inserts)
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 0
+
+
+@pytest.mark.asyncio
+async def test_execute_write_fn_writes_invisible_to_readers_until_task_ends(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    path = str(tmp_path / "test.db")
+    setup_conn = sqlite3.connect(path)
+    setup_conn.execute("create table t (a integer)")
+    setup_conn.close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+
+    def insert_and_check(conn):
+        sqlite_utils.Database(conn)["t"].insert_all({"a": i} for i in range(5))
+        # A separate read-only connection must not see the rows mid-task
+        reader = sqlite3.connect("file:{}?mode=ro".format(path), uri=True)
+        try:
+            return reader.execute("select count(*) from t").fetchone()[0]
+        finally:
+            reader.close()
+
+    mid_task_count = await db.execute_write_fn(insert_and_check)
+    assert mid_task_count == 0
+    # After the task commits the rows are visible
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 5
+
+
 @pytest.mark.parametrize(
     "tables,exists",
     (