#!/usr/bin/env python3 """SQLite statement failures versus transaction rollback. Run: python3 sqlite-rollback-lab.py [--json result.json] All databases are newly created inside a TemporaryDirectory and removed. No server or existing database is accessed. Python uses isolation_level=None; transaction boundaries are explicit SQL. Tested journal mode is DELETE. """ import argparse from contextlib import closing from datetime import datetime import json from pathlib import Path import platform import sqlite3 import tempfile SCHEMA = "CREATE TABLE items (code TEXT NOT NULL PRIMARY KEY, label TEXT NOT NULL)" INSERT = "INSERT INTO items (code, label) VALUES (?, ?)" EXPECTED = { # active at failure, local rows, other-connection rows, final rows "autocommit": (False, ["a"], ["a"], ["a"]), "abort_then_commit": (True, ["a"], [], ["a"]), "abort_then_rollback": (True, ["a"], [], []), "or_rollback": (False, [], [], []), "batch_abort": (True, ["seed"], [], ["seed"]), "batch_fail": (True, ["b", "seed"], [], ["b", "seed"]), "with_exception_out": (True, ["a"], [], []), "with_exception_caught": (True, ["a"], [], ["a"]), } def rows(connection): return [r[0] for r in connection.execute("SELECT code FROM items ORDER BY code").fetchall()] def observed_rows(path): with closing(sqlite3.connect(path, isolation_level=None)) as observer: return rows(observer) def create_connection(path): connection = sqlite3.connect(path, isolation_level=None) mode = connection.execute("PRAGMA journal_mode=DELETE").fetchone()[0] if mode != "delete": connection.close() raise AssertionError("unexpected journal mode: " + mode) connection.execute(SCHEMA) return connection def snapshot(connection, path, error): return { "error_type": type(error).__name__, "active_at_error": connection.in_transaction, "rows_at_error": rows(connection), "observer_at_error": observed_rows(path), } def capture_failure(connection, path, sql, parameters=()): try: connection.execute(sql, parameters) except sqlite3.IntegrityError as error: return snapshot(connection, path, error) raise AssertionError("statement unexpectedly succeeded") def run_case(case, path): with closing(create_connection(path)) as connection: if case.startswith("with_exception"): record = None try: with connection: connection.execute("BEGIN") connection.execute(INSERT, ("a", "first")) try: connection.execute(INSERT, ("a", "duplicate")) except sqlite3.IntegrityError as error: record = snapshot(connection, path, error) if case == "with_exception_out": raise except sqlite3.IntegrityError: if case != "with_exception_out": raise if record is None: raise AssertionError("expected an IntegrityError") else: if case != "autocommit": connection.execute("BEGIN") if case in ("batch_abort", "batch_fail"): connection.execute(INSERT, ("seed", "previous statement")) strategy = "FAIL" if case == "batch_fail" else "ABORT" record = capture_failure(connection, path, f"INSERT OR {strategy} INTO items (code,label) VALUES (?,?),(?,?)", ("b", "first in statement", "b", "duplicate in statement")) else: connection.execute(INSERT, ("a", "first")) sql = ("INSERT OR ROLLBACK INTO items (code,label) VALUES (?,?)" if case == "or_rollback" else INSERT) record = capture_failure(connection, path, sql, ("a", "duplicate")) if connection.in_transaction: connection.execute("ROLLBACK" if case == "abort_then_rollback" else "COMMIT") record["case"] = case record["final_rows"] = observed_rows(path) record["active_at_end"] = connection.in_transaction expected = EXPECTED[case] actual = (record["active_at_error"], record["rows_at_error"], record["observer_at_error"], record["final_rows"]) if actual != expected or record["active_at_end"]: raise AssertionError(f"{case}: got {actual}, expected {expected}") return record def insert_batch(connection, batch): """Own one top-level transaction: either all rows commit or none do. Caller supplies a connection configured like the lab. This helper deliberately rejects an existing transaction; it does not manage nested units of work. """ if connection.in_transaction: raise ValueError("insert_batch needs a connection with no active transaction") connection.execute("BEGIN") try: connection.executemany(INSERT, batch) connection.execute("COMMIT") except BaseException: if connection.in_transaction: connection.execute("ROLLBACK") raise def check_batch_helper(directory): checks = [] with closing(create_connection(directory / "helper.db")) as connection: insert_batch(connection, []) if rows(connection): raise AssertionError("empty batch changed data") checks.append("empty batch") try: insert_batch(connection, [("a", "first"), ("a", "duplicate")]) except sqlite3.IntegrityError: pass else: raise AssertionError("duplicate batch accepted") if rows(connection) or connection.in_transaction: raise AssertionError("failed batch left changes or a transaction") checks.append("duplicate batch rolled back") insert_batch(connection, [("a", "first"), ("b", "second")]) if observed_rows(directory / "helper.db") != ["a", "b"]: raise AssertionError("valid batch did not commit") checks.append("valid batch committed") connection.execute("BEGIN") connection.execute(INSERT, ("owned", "caller owns this")) try: insert_batch(connection, [("c", "must not run")]) except ValueError: pass else: raise AssertionError("existing transaction accepted") if rows(connection) != ["a", "b", "owned"] or not connection.in_transaction: raise AssertionError("caller transaction was changed") connection.execute("ROLLBACK") checks.append("existing transaction rejected without changing it") return checks def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", type=Path, help="Write the environment and observed results") options = parser.parse_args() with tempfile.TemporaryDirectory(prefix="gumi-sqlite-rollback-") as temporary: directory = Path(temporary) cases = [run_case(case, directory / (case + ".db")) for case in EXPECTED] helper_checks = check_batch_helper(directory) report = { "run_date": datetime.now().astimezone().date().isoformat(), "python": platform.python_version(), "platform": platform.system(), "sqlite": sqlite3.sqlite_version, "journal_mode": "delete", "python_isolation_level": None, "cases": cases, "helper_checks": helper_checks, } print(f"Python {report['python']} on {report['platform']}; SQLite {report['sqlite']}; journal=delete") print("case | active_at_error | local_at_error | observer_at_error | final") for record in cases: print(f"{record['case']} | {record['active_at_error']} | {record['rows_at_error']}" f" | {record['observer_at_error']} | {record['final_rows']}") print(f"All {len(cases)} cases and {len(helper_checks)} batch-helper checks passed.") if options.json: options.json.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print("JSON written:", options.json) if __name__ == "__main__": main()