Download the scripts and tests (ZIP)
Python scripts, tests, and ready-to-run merge, join, and date practice files with expected results. No signup required.
Compare yesterday’s export with today’s export using a stable ID, even if the rows and columns were reordered. This script produces a JSON report of new records, missing records, and individual fields that changed. It refuses duplicate IDs because silently choosing one matching row would make the report ambiguous.
Tested: Python 3.10 on Windows, September 7, 2026. Tests cover added and removed records, a changed field, unchanged rows, reordered columns and rows, leading-zero IDs, quoted commas, UTF-8 with a BOM, invalid records, and refusing to overwrite an input file. The script has no external dependencies or network calls.
The job: reconcile two report snapshots
Imagine an operations team receiving a full export each morning. The question is not whether the two files are byte-identical: a different sort order alone could change the file. The useful question is which records changed, matched by a unique business identifier.
Save this as before.csv in a scratch folder:
id,status,note
001,open,"a,b"
002,closed,old
004,open,same
Save this as after.csv using UTF-8. Both the columns and the rows have a different order:
note,id,status
same,004,open
café,003,new
"a,b",001,closed
Expected findings: 003 was added, 002 was removed, 001 changed from open to closed, and 004 stayed the same. The note a,b is one field rather than two. This sample uses invented IDs and text so that no business records are needed to reproduce the result.
Complete script
Save the following as compare_csv.py next to the input files, or use the copy in the downloadable package’s examples folder.
"""Compare CSV snapshots by one unique text key. Python 3.10+."""
import csv
from pathlib import Path
def read_snapshot(path, key):
rows = {}
with Path(path).open(encoding='utf-8-sig', newline='') as handle:
reader = csv.reader(handle, strict=True)
header = next(reader, None)
if not header or any(not name.strip() for name in header):
raise ValueError(f'{path}: missing or empty column names')
if len(set(header)) != len(header):
raise ValueError(f'{path}: duplicate column names')
if key not in header:
raise ValueError(f'{path}: missing key column {key!r}')
for values in reader:
if len(values) != len(header):
raise ValueError(f'{path}: wrong field count near line {reader.line_num}')
row = dict(zip(header, values))
identifier = row[key]
if not identifier.strip():
raise ValueError(f'{path}: blank key near line {reader.line_num}')
if identifier in rows:
raise ValueError(f'{path}: duplicate key {identifier!r}')
rows[identifier] = row
return header, rows
def compare_csv(before_path, after_path, key):
before_header, before = read_snapshot(before_path, key)
after_header, after = read_snapshot(after_path, key)
if set(before_header) != set(after_header):
raise ValueError('Column names differ; reconcile the schema before comparing')
fields = sorted(set(before_header) - {key})
changed = []
unchanged = 0
for identifier in sorted(before.keys() & after.keys()):
differences = {
field: {'before': before[identifier][field], 'after': after[identifier][field]}
for field in fields if before[identifier][field] != after[identifier][field]
}
if differences:
changed.append({'key': identifier, 'fields': differences})
else:
unchanged += 1
return {
'key_column': key,
'added': [after[k] for k in sorted(after.keys() - before.keys())],
'removed': [before[k] for k in sorted(before.keys() - after.keys())],
'changed': changed,
'unchanged_count': unchanged,
}
if __name__ == '__main__':
import argparse
import json
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('before')
parser.add_argument('after')
parser.add_argument('--key', required=True)
parser.add_argument('--output', help='Create a new JSON report; never overwrite a file')
args = parser.parse_args()
report = compare_csv(args.before, args.after, args.key)
rendered = json.dumps(report, ensure_ascii=False, indent=2)
if args.output:
with Path(args.output).open('x', encoding='utf-8') as handle:
handle.write(rendered + '\n')
else:
print(rendered)
Run this command from the folder containing the files and script:
python compare_csv.py before.csv after.csv --key id --output changes.json
The command creates changes.json without modifying either source file. With --output, a successful run does not print the report. Omit that option to print JSON in the terminal instead. An existing output causes an error: choose a new report name or deliberately move the existing report before retrying.
Expected report
{
"key_column": "id",
"added": [{"note": "café", "id": "003", "status": "new"}],
"removed": [{"id": "002", "status": "closed", "note": "old"}],
"changed": [
{
"key": "001",
"fields": {"status": {"before": "open", "after": "closed"}}
}
],
"unchanged_count": 1
}
Whitespace in the printed JSON is more expansive because the script uses indentation. Object member order is not the comparison criterion. Added and removed rows contain the complete source record; changed rows contain only the fields whose values differ. If you share the report, remember that it contains source data.
Check the result before using it
A successful comparison exits with code 0 even when records changed. It is not a command that returns a special failure code for differences. In an automated job, first require successful process completion, then parse the new report and inspect added, removed, and changed. An empty changed list alone does not mean the snapshots match: additions or removals can still exist.
For the invented example above, save this as check_report.py beside the report and run python check_report.py without Python’s -O option:
import json
from pathlib import Path
report = json.loads(Path("changes.json").read_text(encoding="utf-8"))
assert report["key_column"] == "id"
assert report["added"] == [{"id": "003", "status": "new", "note": "café"}]
assert report["removed"] == [{"id": "002", "status": "closed", "note": "old"}]
assert report["changed"] == [
{"key": "001", "fields": {"status": {"before": "open", "after": "closed"}}}
]
assert report["unchanged_count"] == 1
assert len(report["removed"]) + len(report["changed"]) + report["unchanged_count"] == 3
assert len(report["added"]) + len(report["changed"]) + report["unchanged_count"] == 3
print("PASS: sample report and both three-row totals match")
The two totals reconcile the three records in each sample: removed + changed + unchanged accounts for the first snapshot, while added + changed + unchanged accounts for the second. These assertions are a sample check, not a general validator for your own exports. Keep real reports tied to their input files and successful run; a JSON file left by an earlier run is not evidence that today’s comparison succeeded.
Reproduce and recover from a duplicate ID
Save this separate synthetic file as bad-after.csv; leave the valid after.csv above intact:
id,status,note
001,open,first
001,closed,second
Use a report name that does not already exist:
python compare_csv.py before.csv bad-after.csv --key id --output rejected.json
Expected: exit code 1, a traceback ending in ValueError: bad-after.csv: duplicate key '001', and no rejected.json created. Input validation happens before the output is opened. The contradictory statuses show why silently keeping the first or last row would hide a data decision.
For real data, ask the export owner to identify the correct unique key or supply an authoritative corrected snapshot. Do not drop a row merely to make the command pass. For this exercise, the original valid after.csv is the corrected sample. Run:
python compare_csv.py before.csv after.csv --key id --output recovered.json
Expected: exit code 0 and the same JSON values as the expected report above. To check the recovered file, change the checker path to recovered.json and rerun it. If you repeat a successful command with the same output path, it fails with FileExistsError and leaves the existing report unchanged. This follows Python’s exclusive creation mode; choose a fresh report path for each run.
Choose the key before interpreting the result
An ID must identify one row within each snapshot. A customer ID will not work for an order export where one customer has several orders; use an order ID or design a composite-key version for that schema. This script deliberately supports one key column. Concatenating unrelated columns without an unambiguous encoding can introduce collisions.
Keys remain strings. 001 and 1 are different identifiers, and whitespace is not trimmed from otherwise nonblank keys. If the export changes IDs between days, an apparent removal and addition may actually be the same business entity. The script cannot infer that relationship. Resolve identity rules upstream.
Also confirm that both files cover the same population and time window. A filtered or partial export makes omitted rows appear removed. “Removed” in the report means absent from the second file, not proof that someone deleted a record in the source application.
Validation is part of the comparison
| Input condition | Behavior |
|---|---|
| Same columns in a different order | Compare by column name |
| Rows sorted differently | Compare by key |
| Repeated ID in either file | Stop; do not pick a winner |
| Missing or blank key | Stop with an explanation |
| Missing, duplicate, or blank column name | Stop |
| Different column sets | Stop until you reconcile the schema |
| Different text in a non-key field | Report old and new values |
The loader explicitly checks field counts. Otherwise a short record could look like an intentional empty value. Parsing behavior is based on the Python csv module; the validation policy is implemented in the example above.
The comparator uses exact text equality. 10, 10.0, and 10.00 are reported as different strings. Dates in different formats also differ. That conservative choice keeps transformations visible. If numeric tolerance, timezone normalization, or ignored fields matter to your workflow, define and test those rules before changing the comparison.
Limits and next steps
Both snapshots and the report are held in memory. Use this example for files that fit comfortably in your environment; no large-file performance result is claimed. The files should remain unchanged while being read. File permissions, invalid encoding, and malformed CSV stop the command. A disk failure during output can leave a partial new JSON report, so successful process completion matters.
For a header-only file, there are no records. Comparing two valid header-only files returns empty change lists. No version of this script automatically updates a database or deletes a record based on the report. Review the diff before using it as input to another system.
Use CSV merging when the files are non-overlapping batches you want to append. Use this comparison when they are snapshots of the same entities. For downloaded files that might be exact copies, use the duplicate-file report.
Original Dadidan Lab example code may be used and adapted, including commercially, without warranty. Report reproducible problems through contact.