Compare two local CSV files using a CLI-selected key column and report keys that are added, removed, or present in both files with changed values, without modifying either file.
Agent and blockchain workflows often export successive CSV snapshots of records such as job definitions or asset metadata. This read-only tool compares two saved snapshots by a chosen column, so you can spot added, removed, and changed records without executing or interpreting cell contents.
Before you start
Python 3.12 is the intended runner; this is not a claim of a tested minimum version.
Uses only Python standard-library `argparse`, `csv`, `json`, `pathlib`, and `sys`.
Requires two readable UTF-8 CSV files with headers and a key column named by `--key`.
Each file must have unique, nonblank key values and consistent row widths.
Both snapshots must have the same column names; column order may differ. Different schemas are rejected with exit 2.
Build and check
Pass the earlier and later CSV paths as positional CLI arguments and supply the exact key header with `--key`.
Open each file as UTF-8 with `newline=''` so the CSV reader handles newline conventions and quoted fields.
Reject missing headers, duplicate header names, a missing key column, malformed CSV, inconsistent row widths, blank keys, and duplicate keys; report errors on stderr and return a nonzero status.
Store records by key and compare shared records field by field, excluding the key column from change detection.
Sort the reported keys and changed field names for stable output, then write one JSON object to stdout; do not print cell values for changed fields.
Limits to understand
Compares only local CSV snapshots and never contacts or tests an external service.
Requires the chosen key to be unique and nonblank in each file; it does not infer identity or merge duplicate records.
Treats all CSV fields as strings and compares them exactly, without type conversion, normalization, or whitespace trimming.
The report identifies changed field names but intentionally omits prior and new field values.
Independent GitHub Actions testing executed six synthetic local cases in CPython 3.12.15: added/removed/changed records, header-only files, reordered columns, duplicate keys, differing header sets and malformed quoting. The stored receipt verifies these cases only; additional checks below are suggested tests, not claims of execution.
Blank keys, short/extra rows, missing files and invalid UTF-8 have rejection paths in the implementation but are not covered by the six saved fixtures. Live services, production data, very large files and resource limits are outside tested scope.
Provenance: stored as an original, AI-assisted OpenViST example with an MIT declaration for this example. Python documentation supports standard-library behavior; it does not establish this example's authorship or licensing. No third-party code license is inferred from those sources.
What this example adds
This is a new local snapshot comparison, not a generic CSV structure check: it matches records by a user-selected key, detects duplicate or blank keys instead of silently overwriting records, and reports changed column names without printing their values. It validates both inputs before producing the JSON result.
Correction: reject differing header sets before comparison and enable strict CSV parsing; code changed and must pass new independent tests.
Run it
Save the code as `compare_csv_records.py`, then run `python3 compare_csv_records.py before.csv after.csv --key id`. Use `-h` for the CLI help. The intended runner is Python 3.12; no tested minimum version is claimed.
This browser preview builds code locally. It does not contact an API or run the code.
Default example ready.
Python example
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 OpenViST contributors
import argparse
import csv
import json
import pathlib
import sys
def read_snapshot(path, key_name):
try:
with path.open('r', encoding='utf-8', newline='') as source:
reader = csv.DictReader(source, strict=True)
headers = reader.fieldnames
if not headers:
raise ValueError(f'{path}: missing CSV header')
if len(headers) != len(set(headers)):
raise ValueError(f'{path}: duplicate header names')
if key_name not in headers:
raise ValueError(f'{path}: key column {key_name!r} not found')
records = {}
for line_number, row in enumerate(reader, start=2):
if None in row:
raise ValueError(f'{path}: extra fields on line {line_number}')
if any(value is None for value in row.values()):
raise ValueError(f'{path}: missing fields on line {line_number}')
key = row[key_name]
if key == '':
raise ValueError(f'{path}: blank key on line {line_number}')
if key in records:
raise ValueError(f'{path}: duplicate key {key!r} on line {line_number}')
records[key] = row
return records, headers
except OSError as exc:
raise ValueError(f'{path}: cannot read file: {exc}') from exc
except UnicodeError as exc:
raise ValueError(f'{path}: input is not valid UTF-8: {exc}') from exc
except csv.Error as exc:
raise ValueError(f'{path}: invalid CSV: {exc}') from exc
def main():
parser = argparse.ArgumentParser(
description='Compare two local CSV snapshots by a unique key column.'
)
parser.add_argument('before', help='earlier CSV file')
parser.add_argument('after', help='later CSV file')
parser.add_argument('--key', required=True, help='exact key-column header')
args = parser.parse_args()
try:
before, before_headers = read_snapshot(pathlib.Path(args.before), args.key)
after, after_headers = read_snapshot(pathlib.Path(args.after), args.key)
if set(before_headers) != set(after_headers):
raise ValueError('CSV snapshots must have the same column names')
except ValueError as exc:
print(f'error: {exc}', file=sys.stderr)
return 2
before_keys = set(before)
after_keys = set(after)
added = sorted(after_keys - before_keys)
removed = sorted(before_keys - after_keys)
changed = []
for key in sorted(before_keys & after_keys):
fields = sorted(
name for name in before[key]
if name != args.key and before[key][name] != after[key][name]
)
if fields:
changed.append({'key': key, 'fields': fields})
result = {'added': added, 'removed': removed, 'changed': changed}
json.dump(result, sys.stdout, ensure_ascii=False, sort_keys=True)
sys.stdout.write('\n')
return 0
if __name__ == '__main__':
sys.exit(main())
What to verify after running
Run the shown example and confirm the JSON reports `r3` as added, `r2` as removed, and `status` as changed for `r1`.
Try two header-only files with the same key header and confirm the output contains empty `added`, `removed`, and `changed` arrays.
Try a CSV with a duplicate key, a blank key, a row with too few fields, and a row with too many fields; confirm each is rejected with an error on stderr and no JSON result on stdout.
Try a missing input path and a file containing invalid UTF-8; confirm both fail clearly without a traceback being presented as a successful comparison.
Try a quoted field containing a comma and confirm it is read as one field; verify identical records produce no `changed` entry.
Try different header sets with a shared key and a shared record: expect exit 2 and no traceback. Try the same columns in a different order: compare successfully. Try malformed quoting: expect exit 2.
Request or format preview
Illustrative preview. No provider response, price or balance is fetched.
Code license: MIT. Keep the license notice with your copy.
Example version 0.1.1 · Documentation reviewed 2026-10-12.
Executed 6 synthetic local fixture cases in isolated CPython 3.12.15 on GitHub Actions, without network access. Declared exit codes, output assertions and input integrity checks passed. Live services, installed third-party packages and production behavior remain outside scope. Editorial approval is still required.
Related browser tools
Inspect related data locally in your browser. These tools have their own input rules and limits; they do not execute this recipe.
