Implement SAV to SQLite wide-table adapter

Author: TonisOrmisson

pyproject.toml

@@ -10,7 +10,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 license = {text = "Apache-2.0"}
 authors = [{name = "OpenStatSpec"}]
-dependencies = ["SQLAlchemy>=2.0"]
+dependencies = ["SQLAlchemy>=2.0", "pandas>=2.2", "pyreadstat>=1.3"]
 
 [project.optional-dependencies]
 dev = ["pytest>=8"]

src/openstatspec/api.py

@@ -10,7 +10,11 @@
 
 
 def capabilities() -> Mapping[str, Any]:
-    return CapabilityDeclaration.empty().as_dict()
+    return CapabilityDeclaration(
+        specification="OpenStatSpec strict wide-table SPSS profile (initial)",
+        formats={"SAV": {"import": True, "export": True}, "ZSAV": {"import": True, "export": False}},
+        database_profiles={"sqlite": {"import": True, "export": True}},
+    ).as_dict()
 
 
 def inspect(source: str | Path, /, **options: Any) -> Mapping[str, Any]:

src/openstatspec/core/__init__.py

@@ -26,4 +26,7 @@ def empty(cls) -> "CapabilityDeclaration":
         return cls()
 
     def as_dict(self) -> dict[str, Any]:
-        return {"specification": self.specification, "formats": self.formats, "database_profiles": self.database_profiles, "operations": {"inspect": False, "import_sav": False, "export_sav": False, "validate": False}}
+        return {
+            "specification": self.specification, "formats": self.formats, "database_profiles": self.database_profiles,
+            "operations": {"inspect": True, "import_sav": True, "export_sav": True, "validate": True},
+        }

src/openstatspec/spss/__init__.py

@@ -3,16 +3,16 @@
 from pathlib import Path
 from typing import Any
 
-from ..core import UnsupportedOperationError
+from .sav import export_sav_dataset, import_sav_dataset, inspect_sav
 
 
 def inspect_source(source: str | Path, **options: Any) -> dict[str, Any]:
-    raise UnsupportedOperationError("No SAV/ZSAV profile is implemented; source metadata cannot be guessed.")
+    return inspect_sav(source)
 
 
 def import_dataset(source: str | Path, *, database_url: Any, dataset_id: str, **options: Any) -> dict[str, Any]:
-    raise UnsupportedOperationError("No SAV/ZSAV-to-SQL profile is implemented; no partial import was attempted.")
+    return import_sav_dataset(source=source, database_url=str(database_url), dataset_id=dataset_id)
 
 
 def export_dataset(*, database_url: Any, dataset_id: str, destination: str | Path, **options: Any) -> dict[str, Any]:
-    raise UnsupportedOperationError("No SQL-to-SAV/ZSAV profile is implemented; no lossy export was attempted.")
+    return export_sav_dataset(database_url=str(database_url), dataset_id=dataset_id, destination=destination)

src/openstatspec/spss/sav.py

@@ -0,0 +1,108 @@
+"""Initial unencrypted SAV/ZSAV adapter using pyreadstat."""
+
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+import pyreadstat
+
+from ..core import LossReport, UnsupportedOperationError
+from ..sql.wide import create_wide_dataset, physical_name, read_wide_dataset
+
+
+def _variables(meta: Any, names: list[str]) -> list[dict[str, Any]]:
+    used = {"__case_ordinal"}
+    labels = dict(getattr(meta, "column_names_to_labels", {}) or {})
+    formats = dict(getattr(meta, "original_variable_types", {}) or {})
+    measures = dict(getattr(meta, "variable_measure", {}) or {})
+    displays = dict(getattr(meta, "variable_display_width", {}) or {})
+    value_labels = dict(getattr(meta, "variable_value_labels", {}) or {})
+    missing_ranges = dict(getattr(meta, "missing_ranges", {}) or {})
+    widths = dict(getattr(meta, "variable_storage_width", {}) or {})
+    result = []
+    for ordinal, source_name in enumerate(names, start=1):
+        kind = "string" if str(formats.get(source_name, "")).upper().startswith("A") else "numeric"
+        result.append({
+            "ordinal": ordinal, "source_name": source_name,
+            "physical_name": physical_name(source_name, used), "storage_kind": kind,
+            "string_width": widths.get(source_name) if kind == "string" else None,
+            "label": labels.get(source_name, "") or "", "format": formats.get(source_name),
+            "measure": measures.get(source_name), "display_width": displays.get(source_name),
+            "value_labels": json.dumps(value_labels.get(source_name, {}), sort_keys=True),
+            "missing_ranges": json.dumps(missing_ranges.get(source_name, []), default=str),
+        })
+    return result
+
+
+def _rows(frame: Any, variables: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    result = []
+    for input_row in frame.to_dict(orient="records"):
+        output_row = {}
+        for item in variables:
+            value = input_row[item["source_name"]]
+            if item["storage_kind"] == "numeric" and isinstance(value, float) and math.isnan(value):
+                value = None
+            elif item["storage_kind"] == "string" and value is None:
+                value = ""
+            output_row[item["physical_name"]] = value
+        result.append(output_row)
+    return result
+
+
+def inspect_sav(source: str | Path) -> dict[str, Any]:
+    source_path = Path(source)
+    _require_source(source_path)
+    _, meta = pyreadstat.read_sav(source_path, metadataonly=True, user_missing=True)
+    return {
+        "source_format": source_path.suffix[1:].upper(), "source_name": source_path.name,
+        "variable_count": len(meta.column_names), "variables": _variables(meta, list(meta.column_names)),
+    }
+
+
+def import_sav_dataset(*, source: str | Path, database_url: str, dataset_id: str) -> dict[str, Any]:
+    source_path = Path(source)
+    _require_source(source_path)
+    frame, meta = pyreadstat.read_sav(source_path, user_missing=True, disable_datetime_conversion=True)
+    variables = _variables(meta, list(frame.columns))
+    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),
+        variables=variables, file_label=getattr(meta, "file_label", "") or "",
+    )
+    return {**result, "loss_report": LossReport().events}
+
+
+def export_sav_dataset(*, database_url: str, dataset_id: str, destination: str | Path) -> dict[str, Any]:
+    destination_path = Path(destination)
+    if destination_path.suffix.lower() != ".sav":
+        raise UnsupportedOperationError("This initial profile exports SAV only; ZSAV output is not yet supported.")
+    dataset, variables, rows = read_wide_dataset(database_url=database_url, dataset_id=dataset_id)
+    import pandas as pd
+    frame = pd.DataFrame(
+        [{item["source_name"]: row[item["physical_name"]] for item in variables} for row in rows],
+        columns=[item["source_name"] for item in variables],
+    )
+    labels = {item["source_name"]: item["label"] for item in variables if item["label"]}
+    formats = {item["source_name"]: item["format"] for item in variables if item["format"]}
+    measures = {item["source_name"]: item["measure"] for item in variables if item["measure"]}
+    displays = {item["source_name"]: item["display_width"] for item in variables if item["display_width"]}
+    value_labels = {
+        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"] != "{}"
+    }
+    pyreadstat.write_sav(
+        frame, destination_path, file_label=dataset["file_label"], column_labels=labels,
+        variable_format=formats, variable_measure=measures,
+        variable_display_width=displays, variable_value_labels=value_labels,
+    )
+    return {
+        "dataset_id": dataset_id, "destination": str(destination_path),
+        "loss_report": ({"code": "missing-ranges-not-exported", "detail": "The initial writer does not preserve user-missing ranges."},),
+    }
+
+
+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/__init__.py

@@ -2,8 +2,8 @@
 
 from typing import Any
 
-from ..core import UnsupportedOperationError
+from .wide import validate_wide_dataset
 
 
 def validate_dataset(*, database_url: Any, dataset_id: str, **options: Any) -> dict[str, Any]:
-    raise UnsupportedOperationError("No OpenStatSpec SQL profile is implemented; validation cannot guess a schema.")
+    return validate_wide_dataset(database_url=str(database_url), dataset_id=dataset_id)

src/openstatspec/sql/wide.py

@@ -0,0 +1,106 @@
+"""SQLite reference SQL profile for the strict OpenStatSpec wide-table contract."""
+
+import re
+from collections.abc import Iterable, Mapping
+from typing import Any
+
+from sqlalchemy import BigInteger, Column, Float, Integer, MetaData, String, Table, Text, create_engine, insert, select
+
+_IDENTIFIER = re.compile(r"[^a-zA-Z0-9_]+")
+
+
+def catalog(metadata: MetaData) -> tuple[Table, Table]:
+    datasets = Table(
+        "dataset_catalog", metadata,
+        Column("dataset_id", String(255), primary_key=True),
+        Column("data_table", String(255), nullable=False, unique=True),
+        Column("source_format", String(16), nullable=False),
+        Column("source_name", Text, nullable=False),
+        Column("case_count", BigInteger, nullable=False),
+        Column("file_label", Text, nullable=False, default=""),
+    )
+    variables = Table(
+        "variable_catalog", metadata,
+        Column("dataset_id", String(255), primary_key=True),
+        Column("ordinal", Integer, primary_key=True),
+        Column("source_name", String(255), nullable=False),
+        Column("physical_name", String(255), nullable=False),
+        Column("storage_kind", String(16), nullable=False),
+        Column("string_width", Integer),
+        Column("label", Text, nullable=False, default=""),
+        Column("format", String(64)),
+        Column("measure", String(32)),
+        Column("display_width", Integer),
+        Column("value_labels", Text, nullable=False, default="{}"),
+        Column("missing_ranges", Text, nullable=False, default="[]"),
+    )
+    return datasets, variables
+
+
+def physical_name(source_name: str, used: set[str]) -> str:
+    stem = _IDENTIFIER.sub("_", source_name).strip("_").lower() or "variable"
+    stem = stem[:54]
+    candidate, suffix = stem, 2
+    while candidate.lower() in used or candidate.startswith("__"):
+        candidate = f"{stem[:50]}_{suffix}"
+        suffix += 1
+    used.add(candidate.lower())
+    return candidate
+
+
+def data_table_name(dataset_id: str) -> str:
+    stem = _IDENTIFIER.sub("_", dataset_id).strip("_").lower() or "dataset"
+    return f"data_{stem[:48]}"
+
+
+def create_wide_dataset(
+    *, database_url: str, dataset_id: str, source_name: str, source_format: str,
+    rows: Iterable[Mapping[str, Any]], variables: list[dict[str, Any]], file_label: str = "",
+) -> dict[str, Any]:
+    engine = create_engine(database_url)
+    metadata = MetaData()
+    datasets, variable_catalog = catalog(metadata)
+    data_table = Table(
+        data_table_name(dataset_id), metadata,
+        Column("__case_ordinal", BigInteger, primary_key=True, nullable=False),
+        *(Column(item["physical_name"], Float() if item["storage_kind"] == "numeric" else Text(),
+                 nullable=item["storage_kind"] == "numeric") for item in variables),
+    )
+    with engine.begin() as connection:
+        metadata.create_all(connection, tables=[datasets, variable_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.")
+        data_table.create(connection)
+        materialized = [{"__case_ordinal": ordinal, **row} for ordinal, row in enumerate(rows, start=1)]
+        connection.execute(insert(datasets).values(
+            dataset_id=dataset_id, data_table=data_table.name, source_format=source_format,
+            source_name=source_name, case_count=len(materialized), file_label=file_label,
+        ))
+        connection.execute(insert(variable_catalog), [dict(dataset_id=dataset_id, **item) for item in variables])
+        if materialized:
+            connection.execute(insert(data_table), materialized)
+    return {"dataset_id": dataset_id, "data_table": data_table.name, "case_count": len(materialized)}
+
+
+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)
+    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)
+        variables = connection.execute(
+            select(variable_catalog).where(variable_catalog.c.dataset_id == dataset_id).order_by(variable_catalog.c.ordinal)
+        ).mappings().all()
+        rows = connection.execute(select(data_table).order_by(data_table.c.__case_ordinal)).mappings().all()
+    return dict(dataset), [dict(item) for item in variables], [dict(item) for item in rows]
+
+
+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)
+    if not variables:
+        raise ValueError("A conforming dataset needs at least one source variable.")
+    if [row["__case_ordinal"] for row in rows] != list(range(1, len(rows) + 1)):
+        raise ValueError("Case ordinals are not contiguous source order.")
+    return {"dataset_id": dataset["dataset_id"], "valid": True, "case_count": len(rows), "variable_count": len(variables)}
+

tests/test_public_api.py

@@ -0,0 +1,41 @@
+import sqlite3
+
+import openstatspec
+import pandas as pd
+import pyreadstat
+
+
+def test_sav_round_trip_uses_one_wide_table_and_catalog(tmp_path) -> None:
+    source = tmp_path / "tiny.sav"
+    database_path = tmp_path / "dataset.sqlite"
+    database = f"sqlite:///{database_path}"
+    exported = tmp_path / "roundtrip.sav"
+    pyreadstat.write_sav(
+        pd.DataFrame({"age": [34.0, None], "name": ["Ada", ""]}),
+        source,
+        file_label="Tiny fixture",
+        column_labels={"age": "Age", "name": "Name"},
+        variable_value_labels={"age": {34.0: "thirty-four"}},
+        variable_measure={"age": "scale", "name": "nominal"},
+        variable_format={"age": "F8.0", "name": "A12"},
+    )
+
+    imported = openstatspec.import_sav(source, database_url=database, dataset_id="tiny")
+    assert imported["case_count"] == 2
+    assert imported["data_table"] == "data_tiny"
+    assert openstatspec.validate(database_url=database, dataset_id="tiny")["valid"] is True
+
+    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", "variable_catalog"]
+    assert connection.execute("select __case_ordinal, age, name from data_tiny order by __case_ordinal").fetchall() == [
+        (1, 34.0, "Ada"), (2, None, "")
+    ]
+
+    openstatspec.export_sav(database_url=database, dataset_id="tiny", destination=exported)
+    frame, meta = pyreadstat.read_sav(exported, user_missing=True)
+    assert frame["age"].iloc[0] == 34.0
+    assert pd.isna(frame["age"].iloc[1])
+    assert frame["name"].tolist() == ["Ada", ""]
+    assert meta.column_labels == ["Age", "Name"]
+