Updated hashing algorithm, made streaming more efficient (hopefully), also refactored lots of code (hopefully nothing is broken)

Author: datsom1

db_diff/__init__.py

@@ -1,10 +1,12 @@
 import csv
-from dictdiffer import diff
 import json
-import hashlib
-
+import xxhash
+from dictdiffer import diff
 
 def load_csv(fp, key=None, dialect=None):
+    """
+    Load a CSV file into a dictionary keyed by the given column or hash.
+    """
     if dialect is None and fp.seekable():
         # Peek at first 1MB to sniff the delimiter and other dialect details
         peek = fp.read(1024**2)
@@ -14,35 +16,35 @@ def load_csv(fp, key=None, dialect=None):
         except csv.Error:
             # Oh well, we tried. Fallback to the default.
             pass
-    fp = csv.reader(fp, dialect=(dialect or "excel"))
-    headings = next(fp)
-    rows = [dict(zip(headings, line)) for line in fp]
+    reader = csv.reader(fp, dialect=(dialect or "excel"))
+    headings = next(reader)
+    rows = [dict(zip(headings, line)) for line in reader]
     if key:
         keyfn = lambda r: r[key]
     else:
-        keyfn = lambda r: hashlib.sha1(
-            json.dumps(r, sort_keys=True).encode("utf8")
-        ).hexdigest()
+        keyfn = lambda r: xxhash.xxh64(json.dumps(r, sort_keys=True).encode("utf8")).hexdigest()
     return {keyfn(r): r for r in rows}
 
-
 def load_json(fp, key=None):
+    """
+    Load a JSON array of objects into a dictionary keyed by the given column or hash.
+    """
     raw_list = json.load(fp)
-    assert isinstance(raw_list, list)
+    if not isinstance(raw_list, list):
+        raise ValueError("JSON file must contain a list of objects.")
     common_keys = set()
     for item in raw_list:
         common_keys.update(item.keys())
     if key:
         keyfn = lambda r: r[key]
     else:
-        keyfn = lambda r: hashlib.sha1(
-            json.dumps(r, sort_keys=True).encode("utf8")
-        ).hexdigest()
+        keyfn = lambda r: xxhash.xxh64(json.dumps(r, sort_keys=True).encode("utf8")).hexdigest()
     return {keyfn(r): _simplify_json_row(r, common_keys) for r in raw_list}
 
-
 def _simplify_json_row(r, common_keys):
-    # Convert list/dict values into JSON serialized strings
+    """
+    Ensure all rows have the same keys and serialize nested structures.
+    """
     for key, value in r.items():
         if isinstance(value, (dict, tuple, list)):
             r[key] = json.dumps(value)
@@ -51,8 +53,10 @@ def _simplify_json_row(r, common_keys):
             r[key] = None
     return r
 
-
 def compare(previous, current, show_unchanged=False, fields=None, ignorefields=None):
+    """
+    Compare two dictionaries of rows and return a diff summary.
+    """
     result = {
         "added": [],
         "removed": [],
@@ -63,14 +67,13 @@ def compare(previous, current, show_unchanged=False, fields=None, ignorefields=N
     # Have the columns changed?
     previous_columns = set(next(iter(previous.values())).keys())
     current_columns = set(next(iter(current.values())).keys())
-    ignore_columns = None
 
     # Apply fields/ignorefields filtering
     if fields:
         compare_columns = set(fields)
     elif ignorefields:
         compare_columns = previous_columns | current_columns
-        compare_columns = compare_columns - set(ignorefields)
+        compare_columns -= set(ignorefields)
     else:
         compare_columns = previous_columns | current_columns
 
@@ -122,20 +125,17 @@ def compare(previous, current, show_unchanged=False, fields=None, ignorefields=N
                 result["changed"].append(changes)
     return result
 
-
 def streaming_compare_csv(prev_path, curr_path, key, compare_columns=None, ignorefields=None, encoding='utf-8', dialect='excel'):
     """
-    Compare two sorted CSV files by streaming, returning a diff dict.
+    Efficiently compare two sorted CSV files by streaming, returning a diff dict.
     """
-    import csv
     result = {
         "added": [],
         "removed": [],
         "changed": [],
         "columns_added": [],
         "columns_removed": [],
     }
-
     with open(prev_path, newline='', encoding=encoding) as f1, open(curr_path, newline='', encoding=encoding) as f2:
         reader1 = csv.DictReader(f1, dialect=dialect)
         reader2 = csv.DictReader(f2, dialect=dialect)
@@ -158,10 +158,8 @@ def streaming_compare_csv(prev_path, curr_path, key, compare_columns=None, ignor
 
         while prev_row or curr_row:
             if prev_row and curr_row:
-                if key not in prev_row or key not in curr_row:
-                    raise KeyError(f"Key column '{key}' missing in one of the rows.")
-                prev_key = str(prev_row[key])
-                curr_key = str(curr_row[key])
+                prev_key = prev_row[key]
+                curr_key = curr_row[key]
                 if prev_key == curr_key:
                     # Check for changes
                     changed_fields = {
@@ -192,46 +190,40 @@ def streaming_compare_csv(prev_path, curr_path, key, compare_columns=None, ignor
                 curr_row = next(reader2, None)
     return result
 
-
 def human_text(result, key=None, current=None, extras=None):
+    """
+    Render the diff result as a human-readable string.
+    """
     title = []
     summary = []
-    show_headers = sum(1 for key in result if result[key]) > 1
+    show_headers = sum(1 for k in result if result[k]) > 1
     if result["columns_added"]:
-        fragment = "{} {} added".format(
-            len(result["columns_added"]),
-            "column" if len(result["columns_added"]) == 1 else "columns",
-        )
+        fragment = f"{len(result['columns_added'])} {'column' if len(result['columns_added']) == 1 else 'columns'} added"
         title.append(fragment)
         summary.extend(
             [fragment, ""]
-            + ["  {}".format(c) for c in sorted(result["columns_added"])]
+            + [f"  {c}" for c in sorted(result["columns_added"])]
             + [""]
         )
     if result["columns_removed"]:
-        fragment = "{} {} removed".format(
-            len(result["columns_removed"]),
-            "column" if len(result["columns_removed"]) == 1 else "columns",
-        )
+        fragment = f"{len(result['columns_removed'])} {'column' if len(result['columns_removed']) == 1 else 'columns'} removed"
         title.append(fragment)
         summary.extend(
             [fragment, ""]
-            + ["  {}".format(c) for c in sorted(result["columns_removed"])]
+            + [f"  {c}" for c in sorted(result["columns_removed"])]
             + [""]
         )
     if result["changed"]:
-        fragment = "{} rows changed".format(len(result["changed"]))
+        fragment = f"{len(result['changed'])} rows changed"
         title.append(fragment)
         if show_headers:
             summary.append(fragment + "\n")
         change_blocks = []
         for details in result["changed"]:
             block = []
-            block.append("  {}: {}".format(key, details["key"]))
+            block.append(f"  {key}: {details['key']}")
             for field, (prev_value, current_value) in details["changes"].items():
-                block.append(
-                    '    {}: "{}" => "{}"'.format(field, prev_value, current_value)
-                )
+                block.append(f'    {field}: "{prev_value}" => "{current_value}"')
             if extras:
                 current_item = current[details["key"]]
                 block.append(human_extras(current_item, extras))
@@ -241,12 +233,12 @@ def human_text(result, key=None, current=None, extras=None):
                 block = []
                 block.append("    Unchanged:")
                 for field, value in details["unchanged"].items():
-                    block.append('      {}: "{}"'.format(field, value))
+                    block.append(f'      {field}: "{value}"')
                 block.append("")
                 change_blocks.append("\n".join(block))
         summary.append("\n".join(change_blocks))
     if result["added"]:
-        fragment = "{} rows added".format(len(result["added"]))
+        fragment = f"{len(result['added'])} rows added"
         title.append(fragment)
         if show_headers:
             summary.append(fragment + "\n")
@@ -259,7 +251,7 @@ def human_text(result, key=None, current=None, extras=None):
         summary.append("\n\n".join(rows))
         summary.append("")
     if result["removed"]:
-        fragment = "{} rows removed".format(len(result["removed"]))
+        fragment = f"{len(result['removed'])} rows removed"
         title.append(fragment)
         if show_headers:
             summary.append(fragment + "\n")
@@ -273,17 +265,17 @@ def human_text(result, key=None, current=None, extras=None):
         summary.append("")
     return (", ".join(title) + "\n\n" + ("\n".join(summary))).strip()
 
-
 def human_row(row, prefix=""):
-    bits = []
-    for key, value in row.items():
-        bits.append("{}{}: {}".format(prefix, key, value))
-    return "\n".join(bits)
-
+    """
+    Render a row as a human-readable string.
+    """
+    return "\n".join(f"{prefix}{key}: {value}" for key, value in row.items())
 
 def human_extras(row, extras):
-    bits = []
-    bits.append("  extras:")
+    """
+    Render extra fields for a row.
+    """
+    bits = ["  extras:"]
     for key, fmt in extras:
-        bits.append("    {}: {}".format(key, fmt.format(**row)))
+        bits.append(f"    {key}: {fmt.format(**row)}")
     return "\n".join(bits)