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

Python project: validate a JSONL batch before one SQLite commit

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

A batch importer can parse and validate every record before publishing the accepted set in one database transaction.

Download Python source kit

Operation contract

The project caps the already received text, rejects duplicate JSON names, checks each receipt schema and ID uniqueness, then inserts the staged rows inside one transaction. A second batch repeats a receipt ID; the database constraint rejects it, and the new batch does not leave a partial row. The first committed receipt remains. Parsing and database publication are separate boundaries.

Failure and ownership boundary

The sample is an owned in-memory connection, not a hosted ingestion service. Production work needs a streaming byte cap, durable database, retry identity, authentication, and crash recovery. Python JSON duplicate keys: reject ambiguous object fields, Python sqlite3 transactions: bind values and roll back failed batches, Python SQLite job project: reject conflicting replays by request identity and Python JSON Lines: cap line bytes and validate the whole batch before returning it are prerequisites.

Working program

python
import json
import sqlite3

def unique_pairs(pairs):
    record = {}
    for key, value in pairs:
        if key in record:
            raise ValueError("duplicate key")
        record[key] = value
    return record

def import_batch(connection, text):
    if type(text) is not str or len(text) > 400:
        raise ValueError("batch size")
    lines = text.splitlines()
    if not 1 <= len(lines) <= 5:
        raise ValueError("batch rows")
    staged = []
    seen = set()
    for line in lines:
        row = json.loads(line, object_pairs_hook=unique_pairs)
        if type(row) is not dict or set(row) != {"id", "minor"}:
            raise ValueError("record shape")
        if type(row["id"]) is not str or not row["id"].startswith("R") or not row["id"][1:].isdigit() or len(row["id"]) > 8:
            raise ValueError("receipt ID")
        if type(row["minor"]) is not int or not 0 <= row["minor"] <= 100000 or row["id"] in seen:
            raise ValueError("amount or duplicate")
        seen.add(row["id"])
        staged.append((row["id"], row["minor"]))
    with connection:
        connection.executemany("INSERT INTO receipt(id, minor) VALUES (?, ?)", staged)

connection = sqlite3.connect(":memory:")
connection.execute("CREATE TABLE receipt(id TEXT PRIMARY KEY, minor INTEGER NOT NULL)")
import_batch(connection, '{"id":"R41","minor":125}')
try:
    import_batch(connection, '{"id":"R42","minor":75}\n{"id":"R41","minor":0}')
except sqlite3.IntegrityError:
    print("batch rolled back")
print(connection.execute("SELECT id, minor FROM receipt ORDER BY id").fetchall())
connection.close()

Output

Output
batch rolled back
[('R41', 125)]

Costs and limits

For n bounded input characters and r rows, parsing and staging use O(n) time and O(n) working storage. SQLite index operations and durable storage costs are additional; an in-memory commit is not a durability test.

Common Mistakes

  • Validate the whole batch before starting publication.
  • A transaction prevents partial inserts here but does not make a changed replay acceptable.

Connected lessons

Python JSON duplicate keys: reject ambiguous object fields, Python sqlite3 transactions: bind values and roll back failed batches, Python SQLite job project: reject conflicting replays by request identity, Python JSON Lines: cap line bytes and validate the whole batch before returning it.

python
validated-jsonl-ledger-project
Storage details