Append-only CSV with deduplication and atomic writes

csv_io.py

"""
csv_io.py — append-only CSV with deduplication and atomic writes.
No pandas required.

Functions:
    read_csv(filepath)                                   → list of dicts
    write_csv(rows, filepath, columns)                   → row count (atomic write)
    append_csv(rows, filepath, columns)                  → None
    append_csv_deduped(rows, filepath, columns, key_fn)  → new row count
"""

import csv
import os
import tempfile
from pathlib import Path


def read_csv(filepath):
    """Read a CSV into a list of dicts. Returns [] if file doesn't exist."""
    filepath = Path(filepath)
    if not filepath.exists():
        return []
    with open(filepath, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def write_csv(rows, filepath, columns):
    """Write rows to CSV atomically (temp file + rename). Creates parent dirs.

    If the process crashes mid-write, the original file is untouched.

    Returns:
        Number of rows written.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=filepath.parent, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore")
            writer.writeheader()
            writer.writerows(rows)
        os.replace(tmp, filepath)
    except BaseException:
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass
        raise
    return len(rows)


def append_csv(rows, filepath, columns):
    """Append rows to CSV, writing header if the file is new."""
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)
    file_exists = filepath.exists()
    with open(filepath, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore")
        if not file_exists:
            writer.writeheader()
        writer.writerows(rows)


def append_csv_deduped(rows, filepath, columns, dedup_key_fn):
    """Append rows to CSV, skipping any whose key already exists in the file.

    Append-only — never rewrites existing rows.

    Args:
        rows:         List of dicts to append.
        filepath:     Path to the CSV file.
        columns:      Column names (used as CSV header).
        dedup_key_fn: Callable(row) -> hashable. Rows with duplicate keys
                      are silently skipped.

    Returns:
        Number of new rows appended.

    Example:
        append_csv_deduped(
            rows, "events.csv", COLUMNS,
            dedup_key_fn=lambda r: (r["date"], r["event_id"]),
        )
    """
    if not rows:
        return 0

    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    file_exists = filepath.exists()
    existing_keys = set()
    if file_exists:
        for row in read_csv(filepath):
            existing_keys.add(dedup_key_fn(row))

    new_rows = []
    for row in rows:
        key = dedup_key_fn(row)
        if key not in existing_keys:
            new_rows.append(row)
            existing_keys.add(key)

    if not new_rows:
        return 0

    with open(filepath, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore", restval="")
        if not file_exists:
            writer.writeheader()
        writer.writerows(new_rows)

    return len(new_rows)