Support sqlite3.connect(autocommit=True) connections

Database() now accepts connections created with the Python 3.12+
autocommit=True option. The library manages transactions itself using
explicit BEGIN/COMMIT/ROLLBACK and savepoint statements, all guarded by
conn.in_transaction, so it behaves identically in driver-level
autocommit mode - the entire test suite passes under
pytest --sqlite-autocommit with no further changes.

autocommit=False connections are still rejected with TransactionError:
in that mode the driver holds an implicit transaction open at all
times, so explicit BEGIN fails and PRAGMA journal_mode cannot run.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PFkrS2nuP8mo1jmT594VCd
This commit is contained in:
Claude 2026-07-09 05:19:01 +00:00
commit 887c6543ee
No known key found for this signature in database
5 changed files with 78 additions and 19 deletions

View file

@ -65,16 +65,75 @@ def test_database_close(tmpdir, memory):
sys.version_info < (3, 12),
reason="sqlite3.connect(autocommit=) requires Python 3.12",
)
@pytest.mark.parametrize("autocommit", [True, False])
def test_autocommit_connections_are_rejected(tmpdir, autocommit):
# These connection modes break commit()/rollback() in ways that
# silently lose data, so the constructor refuses them
conn = sqlite3.connect(str(tmpdir / "test.db"), autocommit=autocommit)
def test_autocommit_false_connections_are_rejected(tmpdir):
# autocommit=False keeps an implicit transaction open at all times,
# which breaks the explicit transaction handling used by every write
# method, so the constructor refuses these connections
conn = sqlite3.connect(str(tmpdir / "test.db"), autocommit=False)
with pytest.raises(TransactionError):
Database(conn)
conn.close()
@pytest.mark.skipif(
sys.version_info < (3, 12),
reason="sqlite3.connect(autocommit=) requires Python 3.12",
)
def test_autocommit_true_connection_writes_persist(tmpdir):
path = str(tmpdir / "test.db")
conn = sqlite3.connect(path, autocommit=True)
db = Database(conn)
db["t"].insert({"id": 1}, pk="id")
db.execute("insert into t (id) values (2)")
db.close()
# The writes survived closing the connection
db2 = Database(path)
assert [r["id"] for r in db2["t"].rows] == [1, 2]
db2.close()
@pytest.mark.skipif(
sys.version_info < (3, 12),
reason="sqlite3.connect(autocommit=) requires Python 3.12",
)
def test_autocommit_true_connection_transactions(tmpdir):
conn = sqlite3.connect(str(tmpdir / "test.db"), autocommit=True)
db = Database(conn)
db["t"].insert({"id": 1}, pk="id")
# atomic() rolls back on error
with pytest.raises(ZeroDivisionError):
with db.atomic():
db["t"].insert({"id": 2}, pk="id")
1 / 0
assert [r["id"] for r in db["t"].rows] == [1]
# atomic() commits on success, including a nested block
with db.atomic():
db["t"].insert({"id": 3}, pk="id")
with db.atomic():
db["t"].insert({"id": 4}, pk="id")
assert [r["id"] for r in db["t"].rows] == [1, 3, 4]
# begin()/rollback() work too
db.begin()
db["t"].insert({"id": 5}, pk="id")
db.rollback()
assert [r["id"] for r in db["t"].rows] == [1, 3, 4]
db.close()
@pytest.mark.skipif(
sys.version_info < (3, 12),
reason="sqlite3.connect(autocommit=) requires Python 3.12",
)
def test_autocommit_true_connection_wal(tmpdir):
conn = sqlite3.connect(str(tmpdir / "test.db"), autocommit=True)
db = Database(conn)
db.enable_wal()
assert db.journal_mode == "wal"
db.disable_wal()
assert db.journal_mode == "delete"
db.close()
@pytest.mark.skipif(
sys.version_info < (3, 12),
reason="sqlite3.LEGACY_TRANSACTION_CONTROL requires Python 3.12",