Validate CSV imports before saving with Python

import csv
import io

text = "id,amount\nA1,1200\nA1,500\nA2,-1\n"
reader = csv.DictReader(io.StringIO(text, newline=""))
if reader.fieldnames != ["id", "amount"]:
    raise ValueError("unexpected headers")

accepted, errors, seen = [], [], set()
for number, row in enumerate(reader, start=2):
    try:
        if None in row or any(value is None for value in row.values()):
            raise ValueError("incorrect field count")
        key = row["id"].strip()
        if not key:
            raise ValueError("empty id")
        amount = int(row["amount"])
        if amount < 0:
            raise ValueError("negative amount")
        if key in seen:
            raise ValueError("duplicate id")
        seen.add(key)
        accepted.append((key, amount))
    except ValueError as error:
        errors.append(f"record {number}: {error}")

print(errors)
print("Ready to commit:", not errors)

Expect duplicate id at record 3, negative amount at record 4 and Ready to commit: False. No database writes occur. Identifiers are case-sensitive under this example’s contract.

Save as csv_import.py. Accepting a text stream supports both real files and StringIO fixtures. Results are structured and the function does not automatically write accepted rows.

# csv_import.py
import csv

def validate_orders(stream):
    reader = csv.DictReader(stream)
    if reader.fieldnames != ["id", "amount"]:
        raise ValueError("expected headers: id,amount")
    accepted, errors, seen = [], [], set()
    for number, row in enumerate(reader, start=2):
        try:
            if None in row or any(v is None for v in row.values()):
                raise ValueError("incorrect field count")
            key = row["id"].strip()
            if not key:
                raise ValueError("empty id")
            amount = int(row["amount"])
            if amount < 0:
                raise ValueError("negative amount")
            if key in seen:
                raise ValueError("duplicate id")
            seen.add(key)
            accepted.append((key, amount))
        except ValueError as error:
            errors.append({"record": number, "message": str(error)})
    return accepted, errors

accepted is an intermediate result. Under all-or-nothing behavior, any errors prevent the commit. Bound file size, error counts and memory for larger imports; this version targets controlled small files.

Place this file beside the validator. The input is validated before opening the database. Inserts then share one transaction, and the primary key also detects duplicates across imports.

# import_orders.py; run only against a disposable demo database.
import csv
import sqlite3
import sys
from csv_import import validate_orders

csv.field_size_limit(1_000_000)
with open(sys.argv[1], encoding="utf-8-sig", newline="") as source:
    rows, errors = validate_orders(source)
if errors:
    for error in errors:
        print(f"record {error['record']}: {error['message']}")
    raise SystemExit(2)

db = sqlite3.connect(sys.argv[2])
try:
    db.execute("CREATE TABLE IF NOT EXISTS orders "
               "(id TEXT PRIMARY KEY, amount INTEGER NOT NULL CHECK(amount>=0))")
    with db:
        db.executemany("INSERT INTO orders(id,amount) VALUES(?,?)", rows)
    print("Imported:", len(rows))
except sqlite3.IntegrityError as error:
    print("Nothing from this transaction was committed:", error)
    raise SystemExit(3)
finally:
    db.close()

Run python import_orders.py orders.csv demo-orders.sqlite. Importing the same file again should exit 3 without partially committing that transaction. This contract rejects duplicates; accepting retries requires a separate batch identity and idempotency policy.

Run python -m pytest -q test_csv_import.py. The final test deliberately makes the second insert violate uniqueness, then checks both preservation of the existing row and absence of the newly attempted row.

# test_csv_import.py
import io
import sqlite3
from csv_import import validate_orders

def test_clean_file():
    rows, errors = validate_orders(io.StringIO("id,amount\nA1,1200\nA2,0\n"))
    assert rows == [("A1", 1200), ("A2", 0)] and not errors

def test_reports_both_errors():
    rows, errors = validate_orders(io.StringIO("id,amount\nA1,1\nA1,2\nA2,-1\n"))
    assert [e["record"] for e in errors] == [3, 4]

def test_rejects_extra_column():
    _, errors = validate_orders(io.StringIO("id,amount\nA1,1,extra\n"))
    assert errors[0]["message"] == "incorrect field count"

def test_database_rolls_back_batch():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE orders(id TEXT PRIMARY KEY,amount INTEGER)")
    with db:
        db.execute("INSERT INTO orders VALUES('existing',1)")
    try:
        with db:
            db.executemany("INSERT INTO orders VALUES(?,?)",
                           [("new", 2), ("existing", 3)])
    except sqlite3.IntegrityError:
        pass
    assert db.execute("SELECT id FROM orders").fetchall() == [("existing",)]
    db.close()

Expect 4 passed. The suite separates file-validation failures from database-constraint failures. Record numbers and repairable messages are generally enough for diagnostics without logging complete private rows.