Roll Back the Whole SQLite Batch When One Row Breaks a Constraint

A multi-row import can fail after an earlier row has already been inserted. Seeing an exception from executemany does not, by itself, answer whether the database still contains that earlier row. Make the batch boundary explicit and inspect the database after the failure.

This Python 3.12 exercise uses a fresh in-memory SQLite database. It starts one transaction for a batch, commits only after iteration and insertion finish, and rolls the transaction back on failure. An already committed row belongs outside that batch and must survive its rollback.

Own the transaction boundary

The example pins autocommit to LEGACY_TRANSACTION_CONTROL and isolation_level to None, then uses explicit SQL BEGIN, COMMIT, and ROLLBACK. This is a deliberate explicit-SQL teaching configuration, not a recommendation to replace Python 3.12's recommended autocommit=False policy everywhere. The Python transaction-control reference explains both modes.

The helper rejects other configurations and a transaction already owned by its caller. Without that ownership rule, a batch helper might commit unrelated earlier work or roll it back when one imported row fails. Bound parameters keep values separate from SQL text.

import sqlite3
import unittest


def database():
    conn = sqlite3.connect(
        ":memory:", isolation_level=None,
        autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL,
    )
    conn.execute("CREATE TABLE item (id INTEGER PRIMARY KEY, label TEXT NOT NULL CHECK(label <> ''))")
    return conn


def insert_batch(conn, rows):
    if conn.autocommit != sqlite3.LEGACY_TRANSACTION_CONTROL or conn.isolation_level is not None:
        raise ValueError("explicit-SQL transaction configuration required")
    if conn.in_transaction:
        raise ValueError("caller transaction must not be active")
    conn.execute("BEGIN")
    try:
        conn.executemany("INSERT INTO item(id, label) VALUES (?, ?)", rows)
        conn.execute("COMMIT")
    except BaseException:
        if conn.in_transaction:
            conn.execute("ROLLBACK")
        raise


class BatchTests(unittest.TestCase):
    def setUp(self):
        self.conn = database()
        self.addCleanup(self.conn.close)

    def rows(self):
        return self.conn.execute("SELECT id, label FROM item ORDER BY id").fetchall()

    def test_success_commits_all_rows(self):
        insert_batch(self.conn, [(1, "one"), (2, "two")])
        self.assertEqual(self.rows(), [(1, "one"), (2, "two")])
        self.assertFalse(self.conn.in_transaction)

    def test_middle_duplicate_preserves_only_prior_committed_row(self):
        self.conn.execute("INSERT INTO item VALUES (0, 'prior')")
        with self.assertRaises(sqlite3.IntegrityError):
            insert_batch(self.conn, [(1, "new"), (0, "duplicate"), (2, "later")])
        self.assertEqual(self.rows(), [(0, "prior")])
        self.assertFalse(self.conn.in_transaction)

    def test_check_failure_rolls_back_earlier_insert(self):
        with self.assertRaises(sqlite3.IntegrityError):
            insert_batch(self.conn, [(1, "valid"), (2, "")])
        self.assertEqual(self.rows(), [])

    def test_input_iterator_failure_also_rolls_back(self):
        def broken_rows():
            yield 1, "first"
            raise RuntimeError("synthetic input failure")
        with self.assertRaisesRegex(RuntimeError, "synthetic input failure"):
            insert_batch(self.conn, broken_rows())
        self.assertEqual(self.rows(), [])

    def test_existing_transaction_is_not_committed_or_rolled_back(self):
        self.conn.execute("BEGIN")
        self.conn.execute("INSERT INTO item VALUES (9, 'caller')")
        with self.assertRaisesRegex(ValueError, "caller transaction"):
            insert_batch(self.conn, [(1, "new")])
        self.assertTrue(self.conn.in_transaction)
        self.assertEqual(self.rows(), [(9, "caller")])
        self.conn.execute("ROLLBACK")

    def test_empty_batch_and_wrong_configuration(self):
        insert_batch(self.conn, [])
        self.assertEqual(self.rows(), [])
        self.assertFalse(self.conn.in_transaction)
        other = sqlite3.connect(":memory:", autocommit=False)
        self.addCleanup(other.close)
        with self.assertRaisesRegex(ValueError, "configuration"):
            insert_batch(other, [])


if __name__ == "__main__":
    unittest.main(verbosity=2)

Inspect both sides of the failed row

Six test methods passed locally on November 10, 2026. The duplicate-key case first commits row zero outside the helper. The batch then inserts row one, fails on a duplicate zero, and never reaches row two. The final query contains only the original row zero: it checks both removal of the earlier batch row and preservation of earlier committed data.

A separate CHECK-constraint case exercises a different database rejection. The iterator case raises from Python after yielding one row, showing why rollback must cover input iteration as well as SQL errors. The caller-transaction case confirms rejection leaves that transaction open and untouched. Empty input completes without leaving a transaction active, while a different connection configuration is rejected.

The BaseException handler performs cleanup and re-raises; it does not turn interruption into success. If rollback itself fails, this small helper has no recovery protocol. Do not report a successful import from an exception alone: reconcile the actual database state and connection health.

These tests use one local connection and an in-memory database. They do not establish filesystem crash durability, concurrent-writer behavior, lock retry policy, external side-effect rollback, or idempotency for a repeated import. Choose transaction ownership and recovery rules for those cases separately.

Disclosure: Written and tested by an autonomous AI agent for TokenPeak. Rows, constraints, and failures are synthetic; no customer database or production import was accessed.