Skip to content
AITroveRead. Build. Understand.
Reading list
Make this comfortable

Python SQLite project: transactional batches, duplicate IDs and reopen checks

Last updated: 30 Sept 20264 min read
tutorial
IntermediateBy AITrove Editorial

A SQLite receipt ledger stores a validated batch in one database transaction and checks accepted state after reopening its database file.

Download Python source kit

Operation contract

The batch parser requires positive bounded exact integers and distinct IDs before starting inserts. SQL placeholders bind values. The connection context commits successful work and rolls back a failing batch; a duplicate primary key in a later insert therefore rejects the entire attempted transaction. Reopening the local database verifies that previously accepted receipts are present independently of the original connection.

Failure and ownership boundary

The schema uses a primary key and a positive-amount check as a second boundary. The program owns its temporary directory and explicitly closes connections; the connection context does not itself close a connection. Reopen tests do not simulate a power failure or certify a deployment filesystem’s durability. This project has no authentication, migration runner or remote multi-writer service. Python sqlite3 transactions: bind values and roll back failed batches and Python CSV ingestion: cap bytes, rows and fields before publication feed it.

Working program

python
import sqlite3
import tempfile
from pathlib import Path

def accept_batch(connection, rows):
    if len(rows) > 32:
        raise ValueError("batch budget")
    identifiers = set()
    for receipt_id, amount in rows:
        if type(receipt_id) is not int or type(amount) is not int or not 1 <= receipt_id <= 100000 or not 1 <= amount <= 1000000 or receipt_id in identifiers:
            raise ValueError("invalid receipt batch")
        identifiers.add(receipt_id)
    with connection:
        connection.executemany("INSERT INTO receipts(receipt_id, amount_minor) VALUES (?, ?)", rows)

with tempfile.TemporaryDirectory() as directory:
    database = Path(directory) / "receipts.db"
    connection = sqlite3.connect(database)
    try:
        connection.execute("CREATE TABLE receipts(receipt_id INTEGER PRIMARY KEY, amount_minor INTEGER NOT NULL CHECK(amount_minor > 0))")
        accept_batch(connection, [(41, 125), (42, 75)])
        try:
            accept_batch(connection, [(43, 90), (41, 20)])
        except sqlite3.IntegrityError:
            print("duplicate transaction rolled back")
    finally:
        connection.close()
    reopened = sqlite3.connect(database)
    try:
        print(reopened.execute("SELECT receipt_id, amount_minor FROM receipts ORDER BY receipt_id").fetchall())
    finally:
        reopened.close()

Output

Output
duplicate transaction rolled back
[(41, 125), (42, 75)]

Costs and limits

Validation uses O(r) IDs for r rows; insertion cost depends on the database indexes, journaling and filesystem. SELECT allocates the displayed result rows. No constant database write latency is claimed.

Common Mistakes

  • A connection context manages transaction outcome, not connection closing.
  • A clean reopen test does not prove power-loss recovery.

Connected lessons

Python sqlite3 transactions: bind values and roll back failed batches, Python CSV ingestion: cap bytes, rows and fields before publication, Python file project: stage a report before replacing the visible file, Data & Transactions.

Trace the related workflow

Python SQLite checkpoint project: commit data and cursor together.

python
sqlite-cli-project
Storage details