Persist fidelity events across exports

Author: TonisOrmisson

src/openstatspec/spss/sav.py

@@ -10,7 +10,7 @@
 import pyreadstat
 
 from ..core import LossReport, UnsupportedOperationError
-from ..sql.wide import create_wide_dataset, physical_name, read_wide_dataset
+from ..sql.wide import create_wide_dataset, physical_name, read_fidelity_events, read_wide_dataset
 
 
 def _variables(meta: Any, names: list[str]) -> list[dict[str, Any]]:
@@ -76,6 +76,7 @@ def import_sav_dataset(*, source: str | Path, database_url: str, dataset_id: str
     _require_source(source_path)
     frame, meta = pyreadstat.read_sav(source_path, user_missing=True, disable_datetime_conversion=True)
     variables = _variables(meta, list(frame.columns))
+    import_loss_report = _import_loss_report(meta)
     result = create_wide_dataset(
         database_url=database_url, dataset_id=dataset_id, source_name=source_path.name,
         source_format=source_path.suffix[1:].upper(), rows=_rows(frame, variables),
@@ -88,8 +89,9 @@ def import_sav_dataset(*, source: str | Path, database_url: str, dataset_id: str
         imported_at=datetime.now(UTC).isoformat(),
         documents=json.dumps(list(getattr(meta, "notes", []) or [])),
         multiple_response_sets=json.dumps(dict(getattr(meta, "mr_sets", {}) or {}), default=str),
+        fidelity_events=import_loss_report,
     )
-    return {**result, "loss_report": _import_loss_report(meta)}
+    return {**result, "loss_report": import_loss_report}
 
 
 def export_sav_dataset(*, database_url: str, dataset_id: str, destination: str | Path, allow_loss: tuple[str, ...] = ()) -> dict[str, Any]:
@@ -114,7 +116,10 @@ def export_sav_dataset(*, database_url: str, dataset_id: str, destination: str |
         item["source_name"]: {(float(key) if item["storage_kind"] == "numeric" else key): value for key, value in json.loads(item["value_labels"]).items()}
         for item in variables if item["value_labels"] != "{}"
     }
-    loss_report = _export_loss_report(dataset, variables)
+    loss_report = _merge_loss_reports(
+        read_fidelity_events(database_url=database_url, dataset_id=dataset_id),
+        _export_loss_report(dataset, variables),
+    )
     rejected = [event["code"] for event in loss_report if event["code"] not in allow_loss]
     if rejected:
         raise UnsupportedOperationError("Export requires explicit allow_loss for: " + ", ".join(rejected))
@@ -160,6 +165,16 @@ def _export_loss_report(dataset: dict[str, Any], variables: list[dict[str, Any]]
         events.append({"code": "variable-alignment-not-exported", "detail": "The SAV writer has no variable-alignment output capability."})
     return tuple(events)
 
+
+def _merge_loss_reports(*reports: tuple[dict[str, str], ...]) -> tuple[dict[str, str], ...]:
+    """Keep one deterministic diagnostic per loss code across import and export."""
+    events: dict[str, dict[str, str]] = {}
+    for report in reports:
+        for event in report:
+            events.setdefault(event["code"], event)
+    return tuple(events[code] for code in sorted(events))
+
+
 def _require_source(source_path: Path) -> None:
     if source_path.suffix.lower() not in {".sav", ".zsav"}:
         raise UnsupportedOperationError("Only unencrypted SAV and ZSAV sources are supported.")

src/openstatspec/sql/wide.py

@@ -11,7 +11,7 @@
 _IDENTIFIER = re.compile(r"[^a-zA-Z0-9_]+")
 
 
-def catalog(metadata: MetaData) -> tuple[Table, Table]:
+def catalog(metadata: MetaData) -> tuple[Table, Table, Table]:
     datasets = Table(
         "dataset_catalog", metadata,
         Column("dataset_id", String(255), primary_key=True),
@@ -46,8 +46,13 @@ def catalog(metadata: MetaData) -> tuple[Table, Table]:
         Column("value_labels", Text, nullable=False, default="{}"),
         Column("missing_ranges", Text, nullable=False, default="[]"),
     )
-    return datasets, variables
-
+    fidelity_events = Table(
+        "fidelity_event_catalog", metadata,
+        Column("dataset_id", String(255), primary_key=True),
+        Column("code", String(128), primary_key=True),
+        Column("detail", Text, nullable=False),
+    )
+    return datasets, variables, fidelity_events
 
 
 def multiple_response_set_catalog(metadata: MetaData) -> Table:
@@ -103,12 +108,13 @@ def create_wide_dataset(
     source_created_at: str | None = None, source_modified_at: str | None = None,
     imported_at: str = "",
     multiple_response_sets: str = "{}",
+    fidelity_events: Iterable[Mapping[str, str]] = (),
 ) -> dict[str, Any]:
     profile = validate_connection_url(database_url)
     preflight(profile, len(variables))
     engine = create_engine(database_url)
     metadata = MetaData()
-    datasets, variable_catalog = catalog(metadata)
+    datasets, variable_catalog, fidelity_event_catalog = catalog(metadata)
     multiple_response_catalog = multiple_response_set_catalog(metadata)
     data_table = Table(
         data_table_name(dataset_id), metadata,
@@ -117,7 +123,7 @@ def create_wide_dataset(
                  nullable=item["storage_kind"] == "numeric") for item in variables),
     )
     with engine.begin() as connection:
-        metadata.create_all(connection, tables=[datasets, variable_catalog, multiple_response_catalog])
+        metadata.create_all(connection, tables=[datasets, variable_catalog, multiple_response_catalog, fidelity_event_catalog])
         if connection.execute(select(datasets.c.dataset_id).where(datasets.c.dataset_id == dataset_id)).first():
             raise ValueError(f"Dataset {dataset_id!r} already exists; imports never overwrite a dataset.")
         if connection.execute(select(datasets.c.dataset_id).where(datasets.c.data_table == data_table.name)).first():
@@ -138,12 +144,19 @@ def create_wide_dataset(
         mrset_rows = multiple_response_set_rows(dataset_id, multiple_response_sets)
         if mrset_rows:
             connection.execute(insert(multiple_response_catalog), mrset_rows)
+        event_rows = [
+            {"dataset_id": dataset_id, "code": event["code"], "detail": event["detail"]}
+            for event in fidelity_events
+        ]
+        if event_rows:
+            connection.execute(insert(fidelity_event_catalog), event_rows)
         if materialized:
             try:
                 connection.execute(insert(data_table), materialized)
             except Exception:
                 data_table.drop(connection, checkfirst=True)
                 connection.execute(delete(multiple_response_catalog).where(multiple_response_catalog.c.dataset_id == dataset_id))
+                connection.execute(delete(fidelity_event_catalog).where(fidelity_event_catalog.c.dataset_id == dataset_id))
                 connection.execute(delete(variable_catalog).where(variable_catalog.c.dataset_id == dataset_id))
                 connection.execute(delete(datasets).where(datasets.c.dataset_id == dataset_id))
                 connection.commit()
@@ -154,7 +167,7 @@ def create_wide_dataset(
 def read_wide_dataset(*, database_url: str, dataset_id: str) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
     engine = create_engine(database_url)
     metadata = MetaData()
-    datasets, variable_catalog = catalog(metadata)
+    datasets, variable_catalog, _ = catalog(metadata)
     with engine.connect() as connection:
         dataset = connection.execute(select(datasets).where(datasets.c.dataset_id == dataset_id)).mappings().one()
         data_table = Table(dataset["data_table"], MetaData(), autoload_with=connection)
@@ -165,6 +178,21 @@ def read_wide_dataset(*, database_url: str, dataset_id: str) -> tuple[dict[str,
     return dict(dataset), [dict(item) for item in variables], [dict(item) for item in rows]
 
 
+def read_fidelity_events(*, database_url: str, dataset_id: str) -> tuple[dict[str, str], ...]:
+    """Read import-time fidelity diagnostics for a catalogued dataset."""
+    engine = create_engine(database_url)
+    metadata = MetaData()
+    _, _, fidelity_event_catalog = catalog(metadata)
+    with engine.connect() as connection:
+        fidelity_event_catalog.create(connection, checkfirst=True)
+        events = connection.execute(
+            select(fidelity_event_catalog)
+            .where(fidelity_event_catalog.c.dataset_id == dataset_id)
+            .order_by(fidelity_event_catalog.c.code)
+        ).mappings().all()
+    return tuple({"code": item["code"], "detail": item["detail"]} for item in events)
+
+
 def validate_wide_dataset(*, database_url: str, dataset_id: str) -> dict[str, Any]:
     dataset, variables, rows = read_wide_dataset(database_url=database_url, dataset_id=dataset_id)
     profile = validate_connection_url(database_url)

tests/test_loss_reports.py

@@ -27,4 +27,52 @@ def test_export_reports_unwritable_spss_dictionary_features(tmp_path) -> None:
     assert connection.execute("select set_name, member_ordinal, variable_name from multiple_response_set_catalog").fetchall() == [("set1", 1, None)]
     assert {event["code"] for event in result["loss_report"]} == {
         "multiple-response-sets-not-exported", "variable-alignment-not-exported", "unobservable-source-dictionary-features"
-    }
+    }
+
+
+def test_persisted_import_fidelity_events_require_consent_after_reopen(tmp_path) -> None:
+    import pandas as pd
+    import pyreadstat
+    import pytest
+
+    from openstatspec.core import UnsupportedOperationError
+
+    source = tmp_path / "source.sav"
+    database_path = tmp_path / "persisted.sqlite"
+    database = f"sqlite:///{database_path}"
+    blocked = tmp_path / "blocked.sav"
+    approved = tmp_path / "approved.sav"
+    pyreadstat.write_sav(pd.DataFrame({"answer": [1.0]}), source)
+
+    imported = __import__("openstatspec").import_sav(
+        source, database_url=database, dataset_id="persisted"
+    )
+    assert {event["code"] for event in imported["loss_report"]} == {
+        "unobservable-source-dictionary-features"
+    }
+
+    import sqlite3
+    connection = sqlite3.connect(database_path)
+    assert connection.execute(
+        "select dataset_id, code from fidelity_event_catalog"
+    ).fetchall() == [("persisted", "unobservable-source-dictionary-features")]
+    connection.close()
+
+    with pytest.raises(
+        UnsupportedOperationError, match="unobservable-source-dictionary-features"
+    ):
+        __import__("openstatspec").export_sav(
+            database_url=database, dataset_id="persisted", destination=blocked
+        )
+    assert not blocked.exists()
+
+    exported = __import__("openstatspec").export_sav(
+        database_url=database,
+        dataset_id="persisted",
+        destination=approved,
+        allow_loss=["unobservable-source-dictionary-features"],
+    )
+    assert approved.exists()
+    assert {event["code"] for event in exported["loss_report"]} == {
+        "unobservable-source-dictionary-features"
+    }

tests/test_sav_sqlite.py

@@ -33,7 +33,7 @@ def test_sav_round_trip_uses_one_wide_table_and_catalog(tmp_path) -> None:
 
     connection = sqlite3.connect(database_path)
     table_names = [row[0] for row in connection.execute("select name from sqlite_master where type = \"table\" order by name")]
-    assert table_names == ["data_tiny", "dataset_catalog", "multiple_response_set_catalog", "variable_catalog"]
+    assert table_names == ["data_tiny", "dataset_catalog", "fidelity_event_catalog", "multiple_response_set_catalog", "variable_catalog"]
     assert connection.execute("select __case_ordinal, age, name from data_tiny order by __case_ordinal").fetchall() == [
         (1, 34.0, "Ada"), (2, None, "")
     ]