Record source and import timestamps

Author: TonisOrmisson

src/openstatspec/spss/sav.py

@@ -3,6 +3,7 @@
 import hashlib
 import json
 import math
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -74,6 +75,9 @@ def import_sav_dataset(*, source: str | Path, database_url: str, dataset_id: str
         variables=variables, file_label=getattr(meta, "file_label", "") or "",
         source_encoding=getattr(meta, "file_encoding", None),
         source_sha256=_sha256(source_path),
+        source_created_at=_iso_datetime(getattr(meta, "creation_time", None)),
+        source_modified_at=_iso_datetime(getattr(meta, "modification_time", None)),
+        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),
     )
@@ -117,6 +121,10 @@ def export_sav_dataset(*, database_url: str, dataset_id: str, destination: str |
 
 
 
+
+def _iso_datetime(value: Any) -> str | None:
+    return value.isoformat() if value is not None else None
+
 def _sha256(path: Path) -> str:
     digest = hashlib.sha256()
     with path.open("rb") as source:

src/openstatspec/sql/wide.py

@@ -18,6 +18,9 @@ def catalog(metadata: MetaData) -> tuple[Table, Table]:
         Column("source_format", String(16), nullable=False),
         Column("source_name", Text, nullable=False),
         Column("source_sha256", String(64), nullable=False),
+        Column("source_created_at", String(40)),
+        Column("source_modified_at", String(40)),
+        Column("imported_at", String(40), nullable=False),
         Column("source_encoding", String(128)),
         Column("case_count", BigInteger, nullable=False),
         Column("file_label", Text, nullable=False, default=""),
@@ -64,6 +67,8 @@ def create_wide_dataset(
     rows: Iterable[Mapping[str, Any]], variables: list[dict[str, Any]], file_label: str = "",
     source_encoding: str | None = None, documents: str = "[]",
     source_sha256: str = "",
+    source_created_at: str | None = None, source_modified_at: str | None = None,
+    imported_at: str = "",
     multiple_response_sets: str = "{}",
 ) -> dict[str, Any]:
     profile = profile_for_url(database_url)
@@ -87,6 +92,8 @@ def create_wide_dataset(
             dataset_id=dataset_id, data_table=data_table.name, source_format=source_format,
             source_name=source_name, source_encoding=source_encoding, case_count=len(materialized),
             source_sha256=source_sha256,
+            source_created_at=source_created_at, source_modified_at=source_modified_at,
+            imported_at=imported_at,
             file_label=file_label, documents=documents,
             multiple_response_sets=multiple_response_sets,
         ))

tests/test_sav_sqlite.py

@@ -37,6 +37,8 @@ def test_sav_round_trip_uses_one_wide_table_and_catalog(tmp_path) -> None:
     ]
     assert connection.execute("select source_encoding, documents from dataset_catalog").fetchone() == ("UTF-8", json.dumps(["Fixture note"]))
     assert connection.execute("select source_sha256 from dataset_catalog").fetchone() == (hashlib.sha256(source.read_bytes()).hexdigest(),)
+    created_at, modified_at, imported_at = connection.execute("select source_created_at, source_modified_at, imported_at from dataset_catalog").fetchone()
+    assert created_at and modified_at and imported_at
 
     openstatspec.export_sav(database_url=database, dataset_id="tiny", destination=exported)
     frame, meta = pyreadstat.read_sav(exported, user_missing=True)