mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-07-23 09:24:31 +02:00
adfcbb4731Database() now accepts all three Python transaction handling modes and behaves identically in each. For autocommit=False connections - where the driver holds an implicit transaction open at all times - the library tracks transaction ownership itself: - A new _explicit_transaction flag distinguishes transactions opened with begin()/atomic() from the driver's implicit one, exposed as a new db.in_transaction property (conn.in_transaction is always True in this mode) - begin() claims the driver's implicit transaction instead of executing BEGIN, which the driver would reject; BEGIN/COMMIT/ ROLLBACK passed to db.execute() are routed through begin()/commit()/ rollback() - Writes outside a user transaction commit the implicit transaction immediately, preserving the library's auto-commit contract - Row-returning statements outside a transaction fetch eagerly and commit, so the implicit read transaction does not hold a shared lock that blocks writes from other connections - PRAGMA and VACUUM run in temporary driver autocommit mode (ensure_autocommit_on() now flips conn.autocommit), since PRAGMAs are silently ignored and VACUUM refused inside the implicit transaction A new pytest --sqlite-autocommit-false option runs the entire suite in this mode, wired into CI alongside --sqlite-autocommit. Tests that asserted on conn.in_transaction or wrote through db.conn without committing now use the mode-aware library API instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFkrS2nuP8mo1jmT594VCd
93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
import pytest
|
|
from sqlite_utils import Database
|
|
from sqlite_utils.db import TransactionError
|
|
|
|
|
|
@pytest.fixture
|
|
def db_path_tmpdir(tmpdir):
|
|
path = tmpdir / "test.db"
|
|
db = Database(str(path))
|
|
return db, path, tmpdir
|
|
|
|
|
|
def test_enable_disable_wal(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
assert len(tmpdir.listdir()) == 1
|
|
assert "delete" == db.journal_mode
|
|
assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
|
|
db.enable_wal()
|
|
assert "wal" == db.journal_mode
|
|
db["test"].insert({"foo": "bar"})
|
|
assert "test.db-wal" in [f.basename for f in tmpdir.listdir()]
|
|
db.disable_wal()
|
|
assert "delete" == db.journal_mode
|
|
assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
|
|
|
|
|
|
def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
with pytest.raises(TransactionError):
|
|
with db.atomic():
|
|
db["test"].insert({"id": 2}, pk="id")
|
|
db.enable_wal()
|
|
# The atomic() block must have rolled back cleanly and the
|
|
# journal mode must be unchanged
|
|
assert db.journal_mode == "delete"
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|
|
|
|
|
|
def test_disable_wal_inside_transaction_raises(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db.enable_wal()
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
with pytest.raises(TransactionError):
|
|
with db.atomic():
|
|
db["test"].insert({"id": 2}, pk="id")
|
|
db.disable_wal()
|
|
assert db.journal_mode == "wal"
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|
|
|
|
|
|
def test_ensure_autocommit_on(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
previous_isolation_level = db.conn.isolation_level
|
|
previous_autocommit = getattr(db.conn, "autocommit", None)
|
|
with db.ensure_autocommit_on():
|
|
# Driver-level autocommit mode: raw writes on the connection do not
|
|
# open an implicit transaction, they commit immediately
|
|
db.conn.execute("create table t1 (id integer)")
|
|
db.conn.execute("insert into t1 values (1)")
|
|
assert not db.conn.in_transaction
|
|
# Restored afterwards
|
|
assert db.conn.isolation_level == previous_isolation_level
|
|
assert getattr(db.conn, "autocommit", None) == previous_autocommit
|
|
assert db.execute("select count(*) from t1").fetchone()[0] == 1
|
|
|
|
|
|
def test_enable_wal_noop_inside_transaction_is_allowed(db_path_tmpdir):
|
|
# Calling enable_wal() when WAL is already enabled is a no-op,
|
|
# so it is fine inside a transaction
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db.enable_wal()
|
|
with db.atomic():
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
db.enable_wal()
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|
|
|
|
|
|
def test_ensure_autocommit_on_inside_transaction_raises(db_path_tmpdir):
|
|
# Setting isolation_level commits any pending transaction as a side
|
|
# effect, silently breaking the caller's rollback guarantee - so
|
|
# entering autocommit mode with a transaction open is an error
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
db.begin()
|
|
db.execute("insert into test (id) values (2)")
|
|
with pytest.raises(TransactionError):
|
|
with db.ensure_autocommit_on():
|
|
pass
|
|
# The transaction is still open and can still be rolled back
|
|
assert db.in_transaction
|
|
db.rollback()
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|