Compare two CSV snapshots by unique key

Before you begin

Python 3.12 or newer

What this script changesReads source data without editing it. Creates a new destination and refuses overwrite. Rows are held in memory. Spreadsheet formulas in cell text are preserved as text in the CSV; inspect untrusted data before opening in a spreadsheet.

Set up your example

Save the complete code as csv-compare-keys.py in a working folder.
Save the provided sample files in that same folder; run commands from that folder.

Sample files in this download:

sample.csv
lookup.csv
second.csv
wide.csv
numbers.csv
sample.tsv

Download the ZIP, extract it, and open a terminal inside the extracted folder. The source file is csv-compare-keys.py.

Run the script

python "csv-compare-keys.py" --input "sample.csv" --other "second.csv" --key "id" --output "changes.csv"

If your system uses python3 instead of python, use that command for Python examples. Keep paths containing spaces in quotes. Required inputs and optional settings are described in the code and setup steps.

What to expect

Describe added, removed, and changed keyed CSV records. Comparing the samples reports removed ids 1, 2, 3 and added id 4.

The complete script

#!/usr/bin/env python3
'Compare two CSV snapshots by unique key'
import argparse
import json
from pathlib import Path

def write_json(path, value):
    data = json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + "\n"
    with Path(path).open("x", encoding="utf-8", newline="") as handle:
        handle.write(data)

import csv
import random
from collections import Counter, defaultdict
from decimal import Decimal
from datetime import datetime

def read_csv(path, delimiter=","):
    with Path(path).open(encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle, delimiter=delimiter)
        fields = reader.fieldnames
        if not fields or any(not f for f in fields) or len(set(fields)) != len(fields):
            raise ValueError("Use a nonempty, unique header for every column.")
        rows = list(reader)
    if any(None in row or None in row.values() for row in rows):
        raise ValueError("Every row must have exactly the header's number of cells.")
    return fields, rows

def require(fields, names):
    missing = set(names) - set(fields)
    if missing:
        raise ValueError("Missing columns: " + ", ".join(sorted(missing)))

def write_csv(path, fields, rows):
    with Path(path).open("x", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        writer.writerows(rows)

def number(value):
    result = Decimal(value or "0")
    if not result.is_finite():
        raise ValueError("Numbers must be finite.")
    return result

def main():
    parser = argparse.ArgumentParser(description='Compare two CSV snapshots by unique key')
    parser.add_argument("--input", required=True, help="CSV input with a header")
    parser.add_argument("--output", required=True, help="New destination file or directory")
    parser.add_argument("--other", required=True)
    parser.add_argument("--key", required=True)
    args = parser.parse_args()
    fields, rows = read_csv(args.input)
    other_fields, other = read_csv(args.other)
    if fields != other_fields:
        raise ValueError("Both snapshots need the same ordered headers.")
    require(fields, [args.key])
    def index(items):
        mapping = {r[args.key]:r for r in items}
        if len(mapping) != len(items):
            raise ValueError("Snapshot keys must be unique.")
        return mapping
    before, after = index(rows), index(other)
    result_fields = ["key", "change", "field", "before", "after"]
    result = []
    for key in sorted(before.keys() | after.keys()):
        if key not in before or key not in after:
            result.append({"key":key,"change":"added" if key in after else "removed","field":"","before":json.dumps(before.get(key),ensure_ascii=False),"after":json.dumps(after.get(key),ensure_ascii=False)})
        else:
            for field in fields:
                if before[key][field] != after[key][field]:
                    result.append({"key":key,"change":"changed","field":field,"before":before[key][field],"after":after[key][field]})
    write_csv(args.output, result_fields, result)

if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, KeyError, TypeError, ArithmeticError, IndexError, LookupError) as exc:
        raise SystemExit(str(exc))

Open source as plain text · Download source and sample files

How it works

read_csv checks unique headers and row widths before processing.
Unique keys distinguish identity from cell content; changed cells each receive their own difference record.
Comparing the samples reports removed ids 1, 2, 3 and added id 4.

Make it your own

--key

Use a stable identifier with --key; names are suitable only when unique.

Adapt the processing

Edit the field loop to omit volatile fields such as last_seen from cell comparisons.

Change one setting at a time, run it again on sample data, and compare the output. The customization guide walks through this process.

If something goes wrong

Duplicate keys cause an error.

Resolve duplicate identities before comparing so changes cannot be attributed to the wrong record.

The destination already exists.

Choose a new --output path. The script deliberately refuses to replace an existing result.

Reference & next steps

Python 3.12 csv documentation ↗

These official references explain the underlying tools. The Scriptage example and guide are original instructional content. See verification notes for what has been checked and the limits of those checks.

Copies a template for your own notes. It does not send anything.