Use a read-only SELECT transaction for a consistent fixture snapshot

By Batu · English technical notes

Use a read-only SELECT transaction for a consistent fixture snapshot

To keep a SQLite fixture snapshot consistent, run the related SELECT statements inside one explicit transaction. A read transaction continues to see its starting snapshot while another connection commits changes.

The first half of this temporary-database experiment is the failure boundary. Reader and writer are separate connections. The reader fetches A; the writer commits B; then another independent reader query sees both rows. Those two results are individually valid, but they are not one snapshot. The fixture enables WAL mode so the writer can commit while the reader’s transaction is open.

The corrected half deletes B, begins a transaction explicitly on the reader, performs a first SELECT, and lets the writer insert B. The second reader SELECT still returns only A; ROLLBACK ends the read-only use of that transaction. SQLite documents that a connection with an active read transaction sees a historic snapshot until that transaction ends.

isolation_level=None matters here because it disables sqlite3’s implicit transaction handling and lets the code issue BEGIN itself. This use is read-only by convention in this example; BEGIN alone does not grant an operating-system read-only permission or prevent a later write attempt. Python 3.7+ is required because sqlite3.connect() accepts the Path database argument beginning in Python 3.7; sqlite3 itself is standard library.

AI-assistance disclosure: Batu Lab Notes used AI assistance to draft and check this synthetic example.

Sources: SQLite transaction documentation and Python sqlite3 transaction control.

import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory


def labels(con):
    return con.execute("SELECT label FROM items ORDER BY id").fetchall()


with TemporaryDirectory() as directory:
    database = Path(directory) / "fixture.sqlite"
    writer = sqlite3.connect(database, isolation_level=None)
    reader = sqlite3.connect(database, isolation_level=None)
    writer.execute("PRAGMA journal_mode = WAL")
    writer.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, label TEXT)")
    writer.execute("INSERT INTO items(label) VALUES ('A')")

    schema = writer.execute(
        "SELECT sql FROM sqlite_master WHERE name = 'items'"
    ).fetchone()[0]
    naive_first = labels(reader)
    writer.execute("INSERT INTO items(label) VALUES ('B')")
    naive_second = labels(reader)
    assert naive_first == [("A",)]
    assert naive_second == [("A",), ("B",)]

    writer.execute("DELETE FROM items WHERE label = 'B'")
    reader.execute("BEGIN")
    snapshot_first = labels(reader)
    writer.execute("INSERT INTO items(label) VALUES ('B')")
    snapshot_second = labels(reader)
    reader.execute("ROLLBACK")
    assert snapshot_first == [("A",)]
    assert snapshot_second == [("A",)]

    print(f"schema: {schema}")
    print(f"naive first read: {naive_first}")
    print(f"naive second read: {naive_second}")
    print(f"transaction first read: {snapshot_first}")
    print(f"transaction second read: {snapshot_second}")
    reader.close()
    writer.close()

schema: CREATE TABLE items (id INTEGER PRIMARY KEY, label TEXT)
naive first read: [('A',)]
naive second read: [('A',), ('B',)]
transaction first read: [('A',)]
transaction second read: [('A',)]