feat(cms): ingest_log provenance table + writer (refs #594)

src/cms/ingest_log.py

@@ -0,0 +1,128 @@
+"""Append-only DuckDB provenance table for ingest operations.
+
+Complements the JSONL logs in `cms.log`, does not replace them.
+Tracks which reference tables were ingested, when, and by which module.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+
+def new_run_id() -> str:
+    """Generate a new run ID as a UUID4 hex string."""
+    return uuid4().hex
+
+
+def file_sha256(path: str | Path) -> str:
+    """Compute SHA256 hash of a file.
+
+    Args:
+        path: Path to the file to hash.
+
+    Returns:
+        Hexadecimal SHA256 hash of the file contents.
+
+    Raises:
+        FileNotFoundError: If the file does not exist.
+    """
+    path = Path(path)
+    sha256_hash = hashlib.sha256()
+    with open(path, "rb") as f:
+        for chunk in iter(lambda: f.read(4096), b""):
+            sha256_hash.update(chunk)
+    return sha256_hash.hexdigest()
+
+
+def ensure_table(con) -> None:
+    """Create the cms.ingest_log table if it does not exist.
+
+    Args:
+        con: An open DuckDB connection.
+    """
+    con.execute("""
+        CREATE SCHEMA IF NOT EXISTS cms
+    """)
+    con.execute("""
+        CREATE TABLE IF NOT EXISTS cms.ingest_log (
+            run_id VARCHAR,
+            ingested_at TIMESTAMP,
+            module VARCHAR,
+            table_name VARCHAR,
+            rule_id VARCHAR,
+            source_file VARCHAR,
+            sha256 VARCHAR,
+            rows BIGINT,
+            fr_citation VARCHAR,
+            pincite_key VARCHAR
+        )
+    """)
+
+
+def log_ingest(
+    con,
+    *,
+    module: str,
+    table_name: str,
+    rows: int,
+    source_file: str = "",
+    rule_id: str = "",
+    fr_citation: str = "",
+    pincite_key: str = "",
+    run_id: str = "",
+) -> None:
+    """Append an ingest log entry to the provenance table.
+
+    Args:
+        con: An open DuckDB connection.
+        module: Name of the module performing the ingest.
+        table_name: Name of the table being ingested.
+        rows: Number of rows ingested.
+        source_file: Path to the source file (optional). If provided and exists,
+                     its SHA256 hash is computed and stored.
+        rule_id: Rule ID associated with this ingest (optional).
+        fr_citation: Federal Register citation (optional).
+        pincite_key: Pincite key (optional).
+        run_id: Run ID for this ingest operation (optional). If not provided,
+                a new one is generated.
+    """
+    ensure_table(con)
+
+    # Generate run_id if not provided
+    if not run_id:
+        run_id = new_run_id()
+
+    # Compute SHA256 if source_file exists
+    sha256 = ""
+    if source_file:
+        source_path = Path(source_file)
+        if source_path.exists():
+            sha256 = file_sha256(source_path)
+
+    # Get current timestamp in UTC
+    ingested_at = datetime.now(timezone.utc)
+
+    # Insert the row
+    con.execute(
+        """
+        INSERT INTO cms.ingest_log (
+            run_id, ingested_at, module, table_name, rule_id,
+            source_file, sha256, rows, fr_citation, pincite_key
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        [
+            run_id,
+            ingested_at,
+            module,
+            table_name,
+            rule_id,
+            source_file,
+            sha256,
+            rows,
+            fr_citation,
+            pincite_key,
+        ],
+    )

tests/cms/test_ingest_log.py

@@ -0,0 +1,34 @@
+"""Tests for cms.ingest_log — provenance table writer."""
+
+from __future__ import annotations
+
+import duckdb
+
+from cms import ingest_log
+
+
+def test_log_ingest_appends_row(tmp_path):
+    src = tmp_path / "f.csv"
+    src.write_text("a,b\n1,2\n")
+    con = duckdb.connect()
+    ingest_log.log_ingest(
+        con,
+        module="pfs",
+        table_name="pfs.rvu_proposed",
+        rows=2,
+        source_file=str(src),
+        rule_id="CMS-1832-P",
+        fr_citation="90 FR 1",
+        pincite_key="2KVJ2HKX",
+        run_id=ingest_log.new_run_id(),
+    )
+    row = con.execute(
+        "SELECT module, table_name, rows, rule_id, length(sha256) FROM cms.ingest_log"
+    ).fetchone()
+    assert row == ("pfs", "pfs.rvu_proposed", 2, "CMS-1832-P", 64)
+
+
+def test_missing_source_file_logs_empty_hash():
+    con = duckdb.connect()
+    ingest_log.log_ingest(con, module="pfs", table_name="t", rows=0)
+    assert con.execute("SELECT sha256 FROM cms.ingest_log").fetchone()[0] == ""