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
102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
from sqlite_utils import Database
|
|
from sqlite_utils.utils import sqlite3
|
|
import pytest
|
|
|
|
CREATE_TABLES = """
|
|
create table Gosh (c1 text, c2 text, c3 text);
|
|
create table Gosh2 (c1 text, c2 text, c3 text);
|
|
"""
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption(
|
|
"--sqlite-autocommit",
|
|
action="store_true",
|
|
default=False,
|
|
help=(
|
|
"Run every test against connections created with the Python 3.12+ "
|
|
"sqlite3.connect(autocommit=True) mode"
|
|
),
|
|
)
|
|
parser.addoption(
|
|
"--sqlite-autocommit-false",
|
|
action="store_true",
|
|
default=False,
|
|
help=(
|
|
"Run every test against connections created with the Python 3.12+ "
|
|
"sqlite3.connect(autocommit=False) mode"
|
|
),
|
|
)
|
|
|
|
|
|
def pytest_configure(config):
|
|
import sys
|
|
|
|
sys._called_from_test = True # type: ignore[attr-defined]
|
|
|
|
autocommit_true = config.getoption("--sqlite-autocommit")
|
|
autocommit_false = config.getoption("--sqlite-autocommit-false")
|
|
if autocommit_true and autocommit_false:
|
|
raise pytest.UsageError(
|
|
"--sqlite-autocommit and --sqlite-autocommit-false are mutually exclusive"
|
|
)
|
|
if autocommit_true or autocommit_false:
|
|
if sys.version_info < (3, 12):
|
|
raise pytest.UsageError(
|
|
"--sqlite-autocommit and --sqlite-autocommit-false require "
|
|
"Python 3.12 or higher"
|
|
)
|
|
real_connect = sqlite3.connect
|
|
|
|
def autocommit_connect(*args, **kwargs):
|
|
kwargs.setdefault("autocommit", autocommit_true)
|
|
return real_connect(*args, **kwargs)
|
|
|
|
sqlite3.connect = autocommit_connect
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def close_all_databases():
|
|
"""Automatically close all Database objects created during a test."""
|
|
databases = []
|
|
original_init = Database.__init__
|
|
|
|
def tracking_init(self, *args, **kwargs):
|
|
original_init(self, *args, **kwargs)
|
|
databases.append(self)
|
|
|
|
Database.__init__ = tracking_init # type: ignore[method-assign]
|
|
yield
|
|
Database.__init__ = original_init # type: ignore[method-assign]
|
|
for db in databases:
|
|
try:
|
|
db.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def fresh_db():
|
|
return Database(memory=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def existing_db():
|
|
database = Database(memory=True)
|
|
database.executescript("""
|
|
CREATE TABLE foo (text TEXT);
|
|
INSERT INTO foo (text) values ("one");
|
|
INSERT INTO foo (text) values ("two");
|
|
INSERT INTO foo (text) values ("three");
|
|
""")
|
|
return database
|
|
|
|
|
|
@pytest.fixture
|
|
def db_path(tmpdir):
|
|
path = str(tmpdir / "test.db")
|
|
db = sqlite3.connect(path)
|
|
db.executescript(CREATE_TABLES)
|
|
db.commit()
|
|
db.close()
|
|
return path
|