From 84fb4221ab9510f32f6f19c943d7e40c3b6154ec Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 9 Jul 2026 05:50:26 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01N76afGMhBRQk528VF1LTpR --- datasette/database.py | 47 ++++++++++++++-- docs/changelog.rst | 8 +++ docs/internals.rst | 8 ++- tests/test_internals_database.py | 94 ++++++++++++++++++++++++++++++++ 4 files changed, 150 insertions(+), 7 deletions(-) diff --git a/datasette/database.py b/datasette/database.py index e7fe1ed9..e8271c63 100644 --- a/datasette/database.py +++ b/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: diff --git a/docs/changelog.rst b/docs/changelog.rst index e3718543..f58b1c75 100644 --- a/docs/changelog.rst +++ b/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) diff --git a/docs/internals.rst b/docs/internals.rst index 2048c7e4..98fdcd82 100644 --- a/docs/internals.rst +++ b/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 `__ 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. diff --git a/tests/test_internals_database.py b/tests/test_internals_database.py index bad4e8ca..1ca5ee77 100644 --- a/tests/test_internals_database.py +++ b/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", (