Declare SQL target profiles and enforce limits

Author: TonisOrmisson

src/openstatspec/api.py

@@ -6,14 +6,14 @@
 
 from .core import CapabilityDeclaration
 from .spss import export_dataset, import_dataset, inspect_source
-from .sql import validate_dataset
+from .sql import declared_profiles, validate_dataset
 
 
 def capabilities() -> Mapping[str, Any]:
     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}},
+        database_profiles=declared_profiles(),
     ).as_dict()
 
 

src/openstatspec/sql/__init__.py

@@ -2,8 +2,13 @@
 
 from typing import Any
 
+from .profiles import PROFILES
 from .wide import validate_wide_dataset
 
 
+def declared_profiles() -> dict[str, dict[str, object]]:
+    return {profile.name: profile.as_dict() for profile in PROFILES}
+
+
 def validate_dataset(*, database_url: Any, dataset_id: str, **options: Any) -> dict[str, Any]:
     return validate_wide_dataset(database_url=str(database_url), dataset_id=dataset_id)

src/openstatspec/sql/profiles.py

@@ -0,0 +1,54 @@
+"""Dialect capability declarations for the strict wide-table contract.
+
+A declaration is deliberately not a claim that a live target has been tested.
+Importers use this information for preflight checks before creating a dataset.
+"""
+
+from dataclasses import dataclass
+from urllib.parse import urlparse
+
+from ..core import UnsupportedOperationError
+
+
+@dataclass(frozen=True)
+class SqlProfile:
+    name: str
+    url_schemes: tuple[str, ...]
+    max_physical_variables: int
+    identifier_limit: int
+    binary64_numeric: bool
+    lossless_text: bool
+    tested_reference: bool = False
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "max_physical_variables": self.max_physical_variables,
+            "identifier_limit": self.identifier_limit,
+            "binary64_numeric": self.binary64_numeric,
+            "lossless_text": self.lossless_text,
+            "tested_reference": self.tested_reference,
+        }
+
+
+SQLITE = SqlProfile("sqlite", ("sqlite",), 1_999, 255, True, True, True)
+POSTGRESQL = SqlProfile("postgresql", ("postgresql", "postgres"), 1_599, 63, True, True)
+MYSQL = SqlProfile("mysql", ("mysql", "mariadb"), 1_016, 64, True, True)
+PROFILES = (SQLITE, POSTGRESQL, MYSQL)
+
+
+def profile_for_url(database_url: str) -> SqlProfile:
+    scheme = urlparse(database_url).scheme.split("+", 1)[0].lower()
+    for profile in PROFILES:
+        if scheme in profile.url_schemes:
+            return profile
+    raise UnsupportedOperationError(
+        f"No OpenStatSpec SQL profile is declared for database URL scheme {scheme!r}."
+    )
+
+
+def preflight(profile: SqlProfile, variable_count: int) -> None:
+    if variable_count > profile.max_physical_variables:
+        raise UnsupportedOperationError(
+            f"Target capability exceeded: {profile.name} supports at most "
+            f"{profile.max_physical_variables} source variables in one strict wide table."
+        )

src/openstatspec/sql/wide.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from sqlalchemy import BigInteger, Column, Float, Integer, MetaData, String, Table, Text, create_engine, insert, select
+from .profiles import preflight, profile_for_url
 
 _IDENTIFIER = re.compile(r"[^a-zA-Z0-9_]+")
 
@@ -61,6 +62,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 = "[]",
 ) -> dict[str, Any]:
+    profile = profile_for_url(database_url)
+    preflight(profile, len(variables))
     engine = create_engine(database_url)
     metadata = MetaData()
     datasets, variable_catalog = catalog(metadata)