Reuse verified map catalogs and align published SQLite readers

teadata/_sqlite.py

@@ -0,0 +1,11 @@
+"""Read-only SQLite readers with an explicit checkpointed-artifact policy."""
+
+from pathlib import Path
+import sqlite3
+
+
+def readonly_connection(path: str | Path, *, immutable: bool = False) -> sqlite3.Connection:
+    uri = Path(path).resolve().as_uri() + "?mode=ro"
+    if immutable:
+        uri += "&immutable=1"
+    return sqlite3.connect(uri, uri=True)

teadata/boundary_store.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 from typing import Any, Iterable, Optional
 
+from ._sqlite import readonly_connection
 from .manifest import read_sqlite_catalog
 from .release import current_release
 from .teadata_config import canonical_district_number
@@ -62,7 +63,7 @@ def load_boundary(
     if not path:
         return None
     try:
-        with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as conn:
+        with closing(readonly_connection(path, immutable=store_path is None)) as conn:
             for key in _district_lookup_keys(district_number):
                 row = conn.execute(
                     "SELECT wkb FROM boundaries WHERE district_number = ?",

teadata/entity_store.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 from typing import Any, Iterable, Optional
 
+from ._sqlite import readonly_connection
 from .manifest import ContractError, read_sqlite_catalog, read_sqlite_catalog_connection
 from .release import current_release
 from .teadata_config import canonical_district_number
@@ -145,7 +146,7 @@ def load_campus(
 ) -> Optional[dict[str, Any]]:
     """Read one public or source-qualified campus through the canonical store."""
     path = discover_entity_store(store_path)
-    with EntityStore(path) as store:
+    with EntityStore(path, immutable=store_path is None) as store:
         return store.get_campus(campus_number, include_meta=include_meta)
 
 
@@ -174,7 +175,7 @@ def load_district(
     """
 
     try:
-        with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as conn:
+        with closing(readonly_connection(path, immutable=store_path is None)) as conn:
             for key in _district_lookup_keys(district_number):
                 row = conn.execute(query, (key, key)).fetchone()
                 if not row:
@@ -215,7 +216,7 @@ def list_meta_keys(
     if limit:
         query = f"{query} LIMIT {int(limit)}"
     try:
-        with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as conn:
+        with closing(readonly_connection(path, immutable=store_path is None)) as conn:
             rows = conn.execute(query).fetchall()
             keys = [row[0] for row in rows if row and row[0]]
             if keys:

teadata/manifest.py

@@ -15,6 +15,7 @@
 import sqlite3
 from typing import Any, Mapping
 
+from ._sqlite import readonly_connection
 from .dataset_metadata import DataPeriod, DatasetMetadata, MetricDefinition, ResolvedDataset
 from .metric_bindings import MetricBinding, parse_bindings
 from .provenance import ProvenanceStore
@@ -560,10 +561,10 @@ def write_sqlite_catalog(
     )
 
 
-def read_sqlite_catalog(path: str | Path) -> DatasetCatalog:
+def read_sqlite_catalog(path: str | Path, *, immutable: bool = False) -> DatasetCatalog:
     path = Path(path).resolve()
     try:
-        with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)) as connection:
+        with closing(readonly_connection(path, immutable=immutable)) as connection:
             return read_sqlite_catalog_connection(connection)
     except sqlite3.Error as exc:
         raise ContractError(

teadata/map_store.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 from typing import Any, Iterable, Optional, Tuple
 
+from ._sqlite import readonly_connection
 from .manifest import ContractError, read_sqlite_catalog
 from .release import current_release
 from .teadata_config import canonical_district_number
@@ -22,9 +23,13 @@ def map_store_path_for_snapshot(snapshot_path: Path) -> Path:
 
 
 def discover_map_store(explicit: str | Path | None = None) -> Path:
-    path = Path(explicit) if explicit is not None else current_release().artifact("maps")
-    read_sqlite_catalog(path)
-    with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as connection:
+    if explicit is None:
+        # DataRelease owns main-file checksum/catalog validation and its stat cache.
+        path = current_release().artifact("maps")
+    else:
+        path = Path(explicit)
+        read_sqlite_catalog(path)
+    with closing(readonly_connection(path, immutable=explicit is None)) as connection:
         columns = {row[1] for row in connection.execute("PRAGMA table_info(campus_profile)")}
     if "lookup_key" not in columns:
         raise ContractError("Legacy map campus identity schema requires a source-bound rebuild")
@@ -84,7 +89,7 @@ def load_map_payload_parts(
         return None, None
 
     try:
-        with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as conn:
+        with closing(readonly_connection(path, immutable=store_path is None)) as conn:
             for key in _district_lookup_keys(district_number):
                 row = conn.execute(
                     "SELECT payload_base, payload_transfers FROM closures_map WHERE district_number = ?",
@@ -113,7 +118,7 @@ def load_campus_profile_payload(
         return None
     path = discover_map_store(store_path)
     try:
-        with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as conn:
+        with closing(readonly_connection(path, immutable=store_path is None)) as conn:
             row = conn.execute("SELECT payload FROM campus_profile WHERE lookup_key = ?", (key,)).fetchone()
             return _decode_payload(row[0]) if row else None
     except sqlite3.OperationalError as error:

teadata/release.py

@@ -165,7 +165,7 @@ def artifact(self, role: str) -> Path:
         identity = file_identity(path)
         if any(identity[key] != expected[key] for key in ("sha256", "size_bytes")):
             raise ContractError(f"Release artifact checksum or size mismatch: {role}")
-        if role in ROLES - {"snapshot"} and read_sqlite_catalog(path).build_id != self.build_id:
+        if role in ROLES - {"snapshot"} and read_sqlite_catalog(path, immutable=True).build_id != self.build_id:
             raise ContractError(f"Artifact belongs to a different data build: {role}")
         if role in ANALYTICAL_ROLES:
             validate_artifact(path, self.catalog, role, entities_sha256=self._artifacts["entities"]["sha256"])
@@ -247,7 +247,7 @@ def publish_release(
         destination = directory / name
         if (
             role in ROLES - {"snapshot"}
-            and read_sqlite_catalog(source).build_id != catalog.build_id
+            and read_sqlite_catalog(source, immutable=True).build_id != catalog.build_id
         ):
             raise ContractError(f"Cannot publish mixed data builds: {role}")
         if role in ANALYTICAL_ROLES: