Python CSV imports: preserve IDs and reject bad numbers

A CSV export contains customer code 0012 and amount 10.50. If you convert every digit-looking cell to a number, customer 0012 becomes 12. That changes an identifier. CSV readers return text; your import code should decide which fields are IDs, which are numbers, and what a blank field means.

Keep IDs as text

import csv
import io
from decimal import Decimal

source = io.StringIO("customer_id,amount\n0012,10.50\n0013,\n", newline="")
reader = csv.DictReader(source)
for row in reader:
    amount = None if row["amount"] == "" else Decimal(row["amount"])
    print(row["customer_id"], amount)

Validate the shape first

A typo in a header, an extra cell, an invalid amount, or malformed quoting should be reported before writing data. DictReader uses None as the key for extra cells by default. This validator uses strict parsing, but strict mode does not enforce every source-specific CSV convention.

import csv
from decimal import Decimal, InvalidOperation

def clean_rows(file_obj):
    reader = csv.DictReader(file_obj, strict=True)
    try:
        if reader.fieldnames != ["customer_id", "amount"]:
            raise ValueError(f"Expected customer_id,amount; got {reader.fieldnames!r}")
        for record_number, row in enumerate(reader, start=2):
            if None in row or None in row.values():
                raise ValueError(f"Record {record_number}: wrong number of columns")
            customer_id = row["customer_id"]
            if not customer_id.strip():
                raise ValueError(f"Record {record_number}: empty customer_id")
            raw_amount = row["amount"].strip()
            try:
                amount = None if raw_amount == "" else Decimal(raw_amount)
            except InvalidOperation as exc:
                raise ValueError(f"Record {record_number}: invalid amount") from exc
            if amount is not None and not amount.is_finite():
                raise ValueError(f"Record {record_number}: amount must be finite")
            yield customer_id, amount
    except csv.Error as exc:
        raise ValueError(f"Malformed CSV: {exc}") from exc

Check a bad row

import io
try:
    list(clean_rows(io.StringIO("customer_id,amount\n0012,ten\n", newline="")))
except ValueError as exc:
    print(exc)

Keep missing and zero separate

A blank amount is not automatically zero. Decide with the data owner whether it means unknown, not applicable, or invalid. Decimal("0") is a real zero; None signals an absent amount in this example. Do not total missing values until that policy is settled.

Quoted commas and newlines are data

Do not parse CSV with split(",") or splitlines(). A quoted field can contain a comma or a newline, so one logical CSV record may span multiple physical lines. DictReader handles CSV quoting and skips completely blank physical lines. The validator numbers returned records beginning with the header as record 1; skipped blank lines do not increase this count. For a physical source-line reference, reader.line_num reports lines consumed after a successful read; this differs from the record number when records span lines or blank lines are skipped. A parser error may not identify an exact record.

Reject values outside the agreed schema

The function accepts an empty amount as None but rejects a missing column, extra column, empty customer ID, invalid Decimal text, NaN, Infinity, and some malformed quoting. It preserves the customer ID exactly as parsed, including surrounding whitespace; .strip() is used only to test whether it is blank. Strict CSV parsing does not validate a source-specific identifier format. If whitespace or control characters are forbidden, define and check that rule explicitly rather than silently removing them. Completely blank lines are skipped by DictReader; add source-aware validation if they must be rejected. A header-only CSV yields no data rows and no error; reject an empty import explicitly if the business process requires at least one record. Decimal also accepts forms such as 1_0, full-width digits, 1e999999, and -10. A finite Decimal is not necessarily a valid business amount: check permitted characters, scale, sign, and maximum value before accepting it if your schema restricts them. The example also does not enforce unique customer IDs; add that check if the CSV must contain one row per customer.

Test an import before changing data

Start with a small fixture containing a leading-zero ID, a blank amount, a quoted field, an extra cell, and an invalid number. The checks below confirm accepted rows and rejection of bad records. clean_rows() is lazy: it validates a row only when the caller requests it. For a small file, first run validated_rows = list(clean_rows(file_obj)); only after that succeeds should a separate step write those rows. For large files, use a staging table and a transaction or another atomic import strategy instead of keeping every row in memory. Preserve the original file so a failed run can be diagnosed.

import io

file_obj = io.StringIO("customer_id,amount\n0012,10.50\n", newline="")
validated_rows = list(clean_rows(file_obj))
# Only now pass validated_rows to the database write step.

import io
from decimal import Decimal

valid = 'customer_id,amount\n0012,10.50\n0013,\n"00,14",3.20\n 0015 ,0\n'
assert list(clean_rows(io.StringIO(valid, newline=""))) == [
    ("0012", Decimal("10.50")), ("0013", None),
    ("00,14", Decimal("3.20")), (" 0015 ", Decimal("0")),
]

bad_cases = [
    ('customer_id,amount\n0012,2,extra\n', 'wrong number of columns'),
    ('customer_id,amount\n0012,ten\n', 'invalid amount'),
    ('customer_id,amount\n"0012"junk,5\n', 'Malformed CSV'),
]
for source, expected_error in bad_cases:
    try:
        list(clean_rows(io.StringIO(source, newline="")))
    except ValueError as exc:
        assert expected_error in str(exc)
    else:
        raise AssertionError(f"Expected {expected_error}")