CSV Validation in Python: Tested Schema Checks, Input Limits, and Decimal Amounts
This example validates a small customer-import format with four columns: id,email,amount,status. It stops at the first error and yields one validated row at a time. It does not write to a database or promise to make arbitrary spreadsheet files safe.
1. Define the input contract
Open a local UTF-8 file with encoding="utf-8-sig" and newline="". The first setting accepts a UTF-8 byte-order mark; the second lets the CSV parser handle line endings. Commas separate fields and double quotes quote them. Using strict=True catches some malformed quoting; it is not a full RFC 4180 conformance test.
This application requires exactly four unique headers, in any order. Surrounding whitespace is removed from headers and values. IDs are one to eighteen ASCII digits. Amounts have up to twelve integer digits and two fractional digits, with no negative sign or exponent. Status is exactly active, pending, or suspended. The email expression deliberately accepts a limited ASCII subset; it neither implements every legal email address nor verifies mailbox ownership.
2. Bound input, not just the number of accepted rows
A row count alone does not constrain a single huge line. The line iterator reads at most 65,537 decoded characters at once and rejects physical lines longer than 65,536 characters. It also limits the total decoded input to 16,777,216 characters. The standalone command sets the CSV field limit to 65,536 characters. A row containing many columns is still constrained by the input limits.
The validator retains only the current record rather than a list of successes or errors. Memory use still includes text decoding, the current CSV record, and the caller's own state. Calling list(iter_records(...)) accumulates output and defeats that property. The tests use small lists only to compare results. For web uploads, enforce a separate byte limit and timeout before parsing; these character limits are not a substitute for server request limits.
3. Validate headers before mapping values
Consider id,email,amount,status,id. Mapping it directly into a dictionary can hide one ID. Here, duplicate headers fail before any mapping occurs. After trimming and validating the actual header list, each row must have precisely the same number of cells. Reordered headers work because values are mapped against that validated list.
Amounts use a restrictive textual format followed by Decimal. The validator rejects NaN, Infinity, negative values, and excessive precision before conversion. This avoids the previous implementation's float('NaN') comparison loophole. Currency rounding, maximum transaction values, and duplicate customer IDs remain business rules for the receiving application.
4. Treat spreadsheet export as a separate boundary
Parsing a CSV in Python does not execute spreadsheet formulas. Risk arises when untrusted strings are later opened as cells in spreadsheet software. This validator's narrow email format rejects the formula-like example used in the tests, but it is not a general CSV sanitizer. Do not reuse it to declare arbitrary descriptions or names safe.
Choose an export format and escaping strategy for the actual consumer. A CSV writer handles separators and quoting, not universal formula safety. OWASP notes that quote-based mitigations can fail after saving and reopening files. This tutorial makes no claim that prefixing an apostrophe protects every spreadsheet application; it does not modify stored source values with a purported universal sanitizer.
5. Decide what a late failure means
The generator may yield a valid first record and then reject the second. If the caller has already committed the first row, the import is partial. For an all-or-nothing import, consume records within a database transaction and roll it back on validation failure, or validate into a private staging area before committing. For deliberate partial imports, record that policy and report which records were accepted. The command below only counts records, so it performs no database writes.
6. Complete implementation
Requires Python 3.10 or newer; executed checks are documented below. Download csv_validator.py and inspect it before use.
"""Small, fail-fast CSV import example. Python 3.10+, standard library only."""
import csv
import re
from decimal import Decimal
FIELDS = ("id", "email", "amount", "status")
EMAIL = re.compile(r"[A-Za-z0-9._+%-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,63}", re.ASCII)
AMOUNT = re.compile(r"[0-9]{1,12}(?:\.[0-9]{1,2})?", re.ASCII)
def bounded_lines(stream, max_chars, max_line_chars):
"""Bound decoded characters, including physical line terminators."""
used = 0
while True:
line = stream.readline(max_line_chars + 1)
if not line:
return
used += len(line)
if len(line) > max_line_chars:
raise ValueError("Physical line exceeds character limit")
if used > max_chars:
raise ValueError("Input exceeds total character limit")
yield line
def iter_records(stream, *, max_rows=10_000, max_chars=16_777_216,
max_line_chars=65_536):
"""Yield validated rows; fail on the first error without retaining history.
The caller owns the text stream. These are character bounds, not an HTTP
upload-byte limit. Apply byte and time limits before decoding network input.
"""
if min(max_rows, max_chars, max_line_chars) < 1:
raise ValueError("Limits must be positive")
reader = csv.reader(bounded_lines(stream, max_chars, max_line_chars), strict=True)
try:
header = next(reader, None)
if header is None:
raise ValueError("Missing header")
header = [cell.strip() for cell in header]
if len(header) != len(set(header)):
raise ValueError("Duplicate header")
if len(header) != len(FIELDS) or set(header) != set(FIELDS):
raise ValueError("Header must contain exactly id,email,amount,status")
for number, cells in enumerate(reader, 1):
if number > max_rows:
raise ValueError("Row limit exceeded")
if len(cells) != len(header):
raise ValueError(f"Record {number}: wrong column count")
row = dict(zip(header, (cell.strip() for cell in cells)))
if not re.fullmatch(r"[0-9]{1,18}", row["id"], flags=re.ASCII):
raise ValueError(f"Record {number}: invalid id")
if len(row["email"]) > 254 or not EMAIL.fullmatch(row["email"]):
raise ValueError(f"Record {number}: unsupported email format")
if not AMOUNT.fullmatch(row["amount"]):
raise ValueError(f"Record {number}: invalid amount")
if row["status"] not in {"active", "pending", "suspended"}:
raise ValueError(f"Record {number}: invalid status")
yield {"id": int(row["id"]), "email": row["email"],
"amount": Decimal(row["amount"]), "status": row["status"]}
except csv.Error as exc:
raise ValueError(f"CSV parsing failed near physical line {reader.line_num}") from exc
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("path")
args = parser.parse_args()
# This parser setting is process-wide; set once in this standalone CLI.
csv.field_size_limit(65_536)
try:
with open(args.path, encoding="utf-8-sig", newline="") as source:
count = sum(1 for _ in iter_records(source))
except (ValueError, UnicodeError, OSError) as exc:
parser.exit(1, f"Rejected: {exc}\n")
print(f"Validated {count} records; no database writes performed.")
7. Reproduce the checks
Download test_csv_validator.py into the same directory as the validator. Run the following command from that directory:
python -m unittest -v test_csv_validatorThe 15 tests cover exact decimal conversion, reordered and trimmed headers, duplicate/missing/unexpected headers, short and long rows, non-finite and malformed amounts, Unicode digit rejection, a formula-like email, invalid status, unclosed quoting, row and input limits, lazy failure, and a quoted multiline value. Recorded execution results and environments are supplied separately. These checks do not exercise Excel or other spreadsheet programs.
For a real local file, run python csv_validator.py customers.csv. Exit code zero means validation completed; a rejection returns exit code one. The program reports a count rather than printing private customer rows.
8. Limits of this example
This is a focused import example, not a complete upload service. It does not authenticate senders, detect duplicate IDs across files, verify emails, scan malware, or persist transactions. csv.field_size_limit is process-wide; the CLI sets it once, and applications embedding the generator should configure their own parser policy at startup. The input contract intentionally rejects many valid CSV dialects and email formats. Adjust that contract with tests rather than silently accepting more input.