Validate CSV Structure with Python: Exact Headers, Column Order, and Field Counts

An upstream CSV export swaps two columns, but an importer still reads the old positions. An unquoted comma in a note can also shift the remaining values. Being able to open the file does not prove that its structure matches the downstream contract.

This beginner-to-intermediate tutorial builds a Python standard-library command that checks an exact header, exact order, and consistent field counts. It reads one record at a time and exits nonzero on the first problem. A successful JSON report appears only after the entire file passes, making the command useful before a batch import.

1. Method: treat the header as an interface contract

Our example requires sample_id,label,note, in that order:

1. Compare column names exactly, including their order. Added, removed, duplicate, or padded column names are rejected.
2. Every data record must have the same number of fields as the header. Blank lines also fail.
3. Require at least one data record. Empty fields with the correct field count are allowed; data types and business meaning require separate checks.
4. Default to comma-separated fields and double quotes, accepting UTF-8 with or without a BOM. Semicolon-separated input needs an explicit option.

Use csv.reader to obtain field lists. Python’s official CSV documentation explains that DictReader fills missing fields with a default value and stores excess fields in an additional entry. Receiving a dictionary therefore does not prove the field count is correct. This example compares list lengths explicitly.

This check enforces only the stated contract. Follow it with streaming CSV processing and label validation for rules about label values and empty IDs. It is not a comprehensive CSV standards validator.

2. Setup and test files

Use Python 3.9 or newer, with no third-party dependencies. These examples were tested on Linux with Python 3.11.4; shell commands work in Bash or zsh.

1. Create a demo directory and run all subsequent commands there. Choose a new directory name when repeating the exercise:

python3 --version
mkdir csv-schema-demo
cd csv-schema-demo

2. Generate fixtures. valid.csv has a UTF-8 BOM and CRLF line endings, including a quoted comma and a field spanning two lines:

python3 - <<'PY_DATA'
import csv
from pathlib import Path

with open("valid.csv", "w", encoding="utf-8-sig", newline="") as target:
    writer = csv.writer(target)
    writer.writerow(["sample_id", "label", "note"])
    writer.writerow(["001", "1", "quoted, comma"])
    writer.writerow(["002", "0", "two\nlines"])

fixtures = {
    "reordered.csv": "label,sample_id,note\n1,001,ok\n",
    "duplicate.csv": "sample_id,label,label\n001,1,ok\n",
    "short.csv": "sample_id,label,note\n001,1\n",
    "extra.csv": "sample_id,label,note\n001,1,ok,extra\n",
    "broken.csv": 'sample_id,label,note\n001,1,"unfinished\n',
    "blank.csv": "sample_id,label,note\n\n",
    "header-only.csv": "sample_id,label,note\n",
    "empty.csv": "",
    "semicolon.csv": "sample_id;label;note\n001;1;ok\n",
}
for name, text in fixtures.items():
    Path(name).write_text(text, encoding="utf-8")
PY_DATA

This creates files only in the demo directory and reads no production data. For UnicodeDecodeError, start with CSV encoding identification and UTF-8 conversion. Do not discard original bytes with errors="ignore".

3. Complete implementation: validate headers and record widths

Save this code as csv_schema.py in the current directory:

"""Validate an exact CSV header and row widths. Python 3.9+, standard library."""
import argparse
import csv
import json
from pathlib import Path
import sys


def validate(path, columns, delimiter):
    with path.open(encoding="utf-8-sig", newline="") as source:
        reader = csv.reader(source, delimiter=delimiter, strict=True)
        try:
            header = next(reader, None)
            if header != columns:
                raise ValueError(f"header mismatch: expected {columns!r}, got {header!r}")
            records = 0
            for records, row in enumerate(reader, start=1):
                if len(row) != len(columns):
                    raise ValueError(
                        f"record {records}, ending line {reader.line_num}: "
                        f"expected {len(columns)} fields, got {len(row)}"
                    )
            if records == 0:
                raise ValueError("no data records")
        except csv.Error as exc:
            raise ValueError(f"CSV parse error near line {reader.line_num}: {exc}") from exc
    return {"file": str(path), "columns": header, "records": records, "status": "ok"}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("file", type=Path)
    parser.add_argument("--columns", nargs="+", required=True,
                        help="exact column names, in the required order")
    parser.add_argument("--delimiter", choices=[",", ";"], default=",")
    args = parser.parse_args()
    if any(not name or name != name.strip() for name in args.columns):
        parser.error("column names must be nonempty with no surrounding whitespace")
    if len(set(args.columns)) != len(args.columns):
        parser.error("expected column names must be unique")
    try:
        report = validate(args.file, args.columns, args.delimiter)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"ERROR: {args.file}: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(report, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Opening with newline="" leaves CSV newline handling to the parser. strict=True raises an exception for malformed input that the parser recognizes. See the official documentation for csv.reader and Dialect.strict. Strict mode does not enforce your header contract, data types, or every possible quoting mistake, so explicit header and width comparisons are still needed.

The checker leaves the input unchanged and never collects all records into a list. Memory mainly holds the current record and header; an oversized field can still hit the parser’s field-size limit. Expected column names must also be nonempty, unique, and free of surrounding whitespace, so an invalid contract cannot become accepted configuration.

4. Usage and error locations

1. Show help and validate the good file:

python3 csv_schema.py --help
python3 csv_schema.py valid.csv --columns sample_id label note

Expected output:

{"file": "valid.csv", "columns": ["sample_id", "label", "note"], "records": 2, "status": "ok"}

records is 2, rather than the physical line count. The second data record contains a newline in note but remains one CSV record.

2. Check the extra-field fixture and handle this expected failure in the shell:

if python3 csv_schema.py extra.csv --columns sample_id label note; then
  printf 'Schema accepted\n'
else
  printf 'Schema rejected; check stderr above\n'
fi

Standard error includes:

ERROR: extra.csv: record 1, ending line 2: expected 3 fields, got 4

record 1 starts after the header. ending line 2 is the physical ending line consumed by the parser. The official csvreader.line_num documentation distinguishes line counts from record counts. These numbers can differ for multiline records. A parse error’s near line is diagnostic context, not an exact character location.

3. Specify the delimiter for input known to use semicolons:

python3 csv_schema.py semicolon.csv \
  --columns sample_id label note --delimiter ';'

Do not fix records by deleting commas or splitting text with split(","); that breaks valid quoting and multiline fields.

5. Validate success and failure branches

Run this self-check without creating a production failure. It asserts exit codes, error reasons, and successful reports while allowing tested subprocesses to return nonzero:

python3 - <<'PY_CHECK'
import json
import subprocess
import sys

columns = ["--columns", "sample_id", "label", "note"]
cases = [
    (["valid.csv", *columns], 0, None),
    (["reordered.csv", *columns], 1, "header mismatch"),
    (["duplicate.csv", *columns], 1, "header mismatch"),
    (["short.csv", *columns], 1, "expected 3 fields, got 2"),
    (["extra.csv", *columns], 1, "expected 3 fields, got 4"),
    (["broken.csv", *columns], 1, "CSV parse error"),
    (["blank.csv", *columns], 1, "expected 3 fields, got 0"),
    (["header-only.csv", *columns], 1, "no data records"),
    (["empty.csv", *columns], 1, "header mismatch"),
    (["missing.csv", *columns], 1, "ERROR:"),
    (["semicolon.csv", *columns], 1, "header mismatch"),
    (["semicolon.csv", *columns, "--delimiter", ";"], 0, None),
    (["valid.csv", "--columns", "sample_id", "sample_id"], 2, "must be unique"),
]
for arguments, code, message in cases:
    result = subprocess.run(
        [sys.executable, "csv_schema.py", *arguments],
        capture_output=True, text=True, check=False,
    )
    assert result.returncode == code, (arguments, result)
    if code == 0:
        report = json.loads(result.stdout)
        assert report["status"] == "ok", report
        assert report["records"] == (2 if arguments[0] == "valid.csv" else 1), report
        assert result.stderr == "", result
    else:
        assert result.stdout == "" and message in result.stderr, (arguments, result)
print("OK: 13 CSV schema checks")
PY_CHECK

Expected output:

OK: 13 CSV schema checks

The checks cover reordered headers, duplicate column names, short and extra-field records, unclosed quotes, blank lines, header-only input, empty files, missing files, delimiter mismatches, and invalid arguments. The valid fixture also exercises a BOM, CRLF, quoted commas, and embedded newlines so legitimate CSV is accepted.

6. Add a batch gate and troubleshoot

1. Gate downstream work with &&, continuing only when the structure check succeeds:

python3 csv_schema.py valid.csv --columns sample_id label note \
  && python3 -c 'print("Downstream step started")'

The demo downstream command prints one line; replace it with your existing import command. Keep the input unchanged between validation and import. If a producer is still writing, let it finish and validate a fixed output first, otherwise the two steps may read different contents.

2. Fix the source of each problem and validate again:

A correct structure does not prove ID uniqueness. To avoid duplicate imports after merging batches, add CSV duplicate-ID checks with Python and SQLite.

7. Summary

An exact header, column order, and record width catch upstream schema drift and some malformed records before import. Validate with small fixtures containing quotes, multiline fields, and failure cases before adding the check to a batch workflow. Keep structural rules, field-value rules, and ID uniqueness explicit, with meaningful exit codes.