Python CSV: Check Headers Before DictReader Loses Values

A CSV import can contain three header fields and three values while producing a dictionary with only two keys. If the header repeats a name, converting the row into a dictionary merges those positions under one key. Checking the number of keys afterward is too late to recover the overwritten value. Start with the original CSV and inspect its parsed header before building mappings.

This guide uses Python's standard library to separate duplicate names, extra columns, missing columns, and intentionally empty values. It includes a small-file validator that rejects structural problems before returning any rows to a caller. It does not infer the producer's schema, choose an encoding automatically, or establish whether a particular value is valid business data.

Reproduce equal width with fewer dictionary keys

Save each example as a Python file and run it with python3. The examples use synthetic strings and do not modify an existing CSV. This first example preserves the raw fields so the loss is visible independently of the dictionary.

import csv
import io
from collections import Counter

text = 'id,name,name\n0012,given,family\n'
raw = list(csv.reader(io.StringIO(text, newline='')))
header, values = raw
mapped = next(csv.DictReader(io.StringIO(text, newline='')))

assert len(header) == len(values) == 3
assert mapped == {'id': '0012', 'name': 'family'}
duplicates = [name for name, n in Counter(header).items() if n > 1]
assert duplicates == ['name']
print('duplicate names:', duplicates)
print('mapped row:', mapped)

Both fields were parsed; their names collided during mapping. Changing a delimiter, increasing a field-size limit, or reinstalling Python does not create a second dictionary entry for the same name. Retain the original list of fields until the schema check succeeds. If duplicate names are intentional, use positional rows with an agreed schema, or obtain distinct names from the producer. Appending arbitrary suffixes can silently change which value a downstream consumer reads.

Separate extra, missing and empty fields

A unique header does not guarantee that every record has the same width. Diagnose these cases separately: an extra field is outside the header, a missing field was absent from the record, and an empty string was present but empty. They can require different handling in a data pipeline.

import csv
import io

extra = object()
missing = object()
text = 'id,name\n1,A,unexpected\n2\n3,\n'
rows = list(csv.DictReader(
    io.StringIO(text, newline=''),
    restkey=extra,
    restval=missing,
    strict=True,
))
assert rows[0][extra] == ['unexpected']
assert rows[1]['name'] is missing
assert rows[2]['name'] == ''
assert extra not in rows[1] and extra not in rows[2]
print('extra columns, absent value and empty text are distinct')

The sentinel objects are local diagnostic markers. A literal CSV string cannot equal either object by identity, and an actual header cannot collide with the extra object. These dictionaries are not a JSON export format; classify and handle the markers before serialization. A string marker such as MISSING could itself appear in legitimate input.

Notice that strict=True still accepted the ragged records in this example. Parser strictness concerns CSV syntax; it does not enforce your column-count contract. It also does not validate unique names, required names, IDs, date formats or allowed values. Keep syntax errors and schema errors distinct in diagnostics.

Validate before converting records into dictionaries

For a small import, gather the validated rows before making any downstream changes. The function below uses an exact, ordered header contract supplied by the caller. It rejects duplicate and empty names, an unexpected header, blank records, and short or long records. A valid header with no data rows is accepted; add a separate nonempty-import rule if your workflow needs one.

import csv
import io
from collections import Counter


def checked_rows(stream, expected):
    expected = list(expected)
    if (not expected or any(name == '' for name in expected)
            or len(set(expected)) != len(expected)):
        raise ValueError('expected schema must contain unique nonempty names')

    reader = csv.reader(stream, strict=True)
    header = next(reader, None)
    if not header or any(name == '' for name in header):
        raise ValueError('missing or empty header')
    duplicates = [name for name, n in Counter(header).items() if n > 1]
    if duplicates:
        raise ValueError('duplicate header names')
    if header != expected:
        raise ValueError('header differs from expected ordered schema')

    result = []
    for record, row in enumerate(reader, start=2):
        if len(row) != len(header):
            raise ValueError(
                f'width mismatch at record {record}: '
                f'expected {len(header)}, got {len(row)}'
            )
        result.append(dict(zip(header, row)))
    return result


def parse(text):
    return checked_rows(io.StringIO(text, newline=''), ['id', 'name'])


assert parse('id,name\n0012,"Lee, Min"\n') == [
    {'id': '0012', 'name': 'Lee, Min'}
]
assert parse('id,name\n1,"a\nb"\n') == [{'id': '1', 'name': 'a\nb'}]
assert parse('id,name\n1,\n') == [{'id': '1', 'name': ''}]
assert parse('id,name\n') == []

bad_inputs = [
    'id,name,name\n1,A,B\n',
    ',name\n1,A\n',
    '',
    'name,id\nA,1\n',
    'id,name\n1,A,extra\n',
    'id,name\n1\n',
    'id,name\n\n',
    'id,name\n1,"unterminated\n',
]
for text in bad_inputs:
    try:
        parse(text)
    except (ValueError, csv.Error):
        pass
    else:
        raise AssertionError('expected input rejection')

for schema in [[], ['id', 'id'], ['id', '']]:
    try:
        checked_rows(io.StringIO('id,name\n'), schema)
    except ValueError:
        pass
    else:
        raise AssertionError('expected schema rejection')

assert checked_rows(io.StringIO('Name,name\nA,B\n'), ['Name', 'name']) == [
    {'Name': 'A', 'name': 'B'}
]
assert checked_rows(io.StringIO(' id,id\nA,B\n'), [' id', 'id']) == [
    {' id': 'A', 'id': 'B'}
]
print('6 accepted inputs, 8 rejected inputs, 3 rejected schemas')

The record number counts parsed records beginning with the header as record one. It is not a physical line number: a quoted field can span several lines. For physical-line diagnostics, also record reader.line_num when handling an error. Avoid placing sensitive row contents in shared error logs.

Using split(',') instead would break the quoted comma in the first accepted case. Splitting the whole text into lines would also require special treatment for quoted newlines. Let the CSV parser find field boundaries, then check the resulting list. The code preserves strings, including the leading zeros in 0012; numeric conversion belongs after structural validation.

Make normalization and blank-record policies explicit

This validator treats Name and name as different names, and preserves the leading space in id. That is deliberate. If your producer contract requires trimming or case folding, normalize first and then test uniqueness again: two distinct raw names can become one normalized name. Do not silently rename a duplicate or remove a column to make a row fit.

Blank records are rejected here. DictReader normally skips empty rows during iteration, so replacing it with this validator can expose files that previously passed. Decide whether a blank record means damaged input or an allowed separator, document that decision, and test it. A record containing delimiters and empty fields is also different from a completely blank record.

For a real file, open it read-only with newline='' and an explicit encoding agreed with the producer, then pass the open stream to checked_rows. Use the producer's delimiter and quoting settings when adapting the reader. A UTF-8 BOM can affect the first header name; handle a documented BOM at the decoding boundary rather than stripping characters from arbitrary header values. See CSV BOM decoding before DictReader if that is the observed mismatch.

Preserve the validation boundary when imports grow

Returning a list is useful for a small fixture because the caller receives nothing if a later row fails validation. The function can still consume substantial memory, and it does not roll back database writes. Callers must wait for the function to return, validate field meanings, and apply changes under their own transaction or staging policy.

For a large import, an iterator reduces memory but can yield earlier rows before discovering a bad later row. Treating each yielded row as permission to commit immediately would introduce partial imports. Stage the output and promote it after the entire input passes, or use an appropriate transaction that rolls back on parsing, schema and business-rule failures. A second validation pass is insufficient if the file can change between passes; use a stable input snapshot.

Keep a read-only original for diagnosis and compare the final imported field counts and representative values with the intended consumer. The supplied fixtures were run on macOS with Python 3.11.6, 3.13.2 and 3.14.3. They verify parser and validator behavior for these small cases; they do not test Windows, a customer's encoding or dialect, large-file memory use, or database atomicity.