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

Python SQLite checkpoint project: commit data and cursor together

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

A checkpointed import stores its records and last accepted source offset in the same transaction.

Download Python source kit

Operation contract

The owned in-memory database has a receipt table and one checkpoint row. A batch beginning at the expected next offset inserts records, then updates the checkpoint before committing. A later batch with a mismatched starting offset is rejected before writing. The fixture shows the accepted state and that a rejected batch leaves both records and checkpoint unchanged.

Failure and ownership boundary

A source offset alone is not a deduplication key when the upstream can rewrite history or deliver a different payload at the same position. Real importers need source identity, input integrity, replay policy and recovery after a crash. Python sqlite3 transactions: bind values and roll back failed batches, Python SQLite project: transactional batches, duplicate IDs and reopen checks and Python SQLite job project: reject conflicting replays by request identity explain those limits.

Working program

python
import sqlite3

connection = sqlite3.connect(":memory:")
connection.execute("CREATE TABLE receipt(offset INTEGER PRIMARY KEY, identifier TEXT NOT NULL)")
connection.execute("CREATE TABLE checkpoint(id INTEGER PRIMARY KEY CHECK(id=1), last_offset INTEGER NOT NULL)")
connection.execute("INSERT INTO checkpoint VALUES (1, 0)")
connection.commit()

def ingest(start, identifiers):
    if type(start) is not int or start <= 0 or type(identifiers) is not list or not 1 <= len(identifiers) <= 10 or any(type(identifier) is not str or len(identifier) != 3 for identifier in identifiers):
        raise ValueError("bounded import")
    with connection:
        previous = connection.execute("SELECT last_offset FROM checkpoint WHERE id=1").fetchone()[0]
        if start != previous + 1:
            raise ValueError("offset gap or replay")
        for offset, identifier in enumerate(identifiers, start):
            connection.execute("INSERT INTO receipt VALUES (?, ?)", (offset, identifier))
        connection.execute("UPDATE checkpoint SET last_offset=? WHERE id=1", (start + len(identifiers) - 1,))

ingest(1, ["R41", "R42"])
try:
    ingest(4, ["R43"])
except ValueError:
    print("gap rejected")
print("rows:", connection.execute("SELECT offset,identifier FROM receipt ORDER BY offset").fetchall())
print("checkpoint:", connection.execute("SELECT last_offset FROM checkpoint").fetchone()[0])
connection.close()

Output

Output
gap rejected
rows: [(1, 'R41'), (2, 'R42')]
checkpoint: 2

Costs and limits

Each batch inserts k rows with index maintenance and one checkpoint update inside a transaction. The fixture is in memory and sequential; disk durability, writer contention and source acknowledgements require separate tests.

Common Mistakes

  • Do not advance the cursor independently of the committed records.
  • An offset alone cannot prove replay payload identity.

Connected lessons

Python sqlite3 transactions: bind values and roll back failed batches, Python SQLite project: transactional batches, duplicate IDs and reopen checks, Python SQLite job project: reject conflicting replays by request identity.

python
sqlite-checkpoint-project
Storage details