Merge pull request #228 from datakind/feat/remove-custom-institutions

feat: remove custom institutions, enforce PDP / Edvise / Legacy in API and uploads

Author: vishpillai123

src/webapp/databricks.py

@@ -13,19 +13,16 @@
 )
 from google.cloud import storage
 from google.api_core import exceptions as gcs_errors
-from .validation_extension import generate_extension_schema
 from .config import databricks_vars, gcs_vars
 from .utilities import databricksify_inst_name, SchemaType
 from typing import List, Any, Dict, Optional
-from fastapi import HTTPException
 import requests
 import hashlib
 import json
 import gzip
 from cachetools import TTLCache
 import threading
 import re
-import pandas as pd
 
 # Setting up logger
 LOGGER = logging.getLogger(__name__)
@@ -623,78 +620,3 @@ def matches_one(pat: Any) -> bool:
                     return key
 
         return None
-
-    def create_custom_schema_extension(
-        self,
-        bucket_name: str,
-        inst_query: Any,
-        file_name: str,
-        base_schema: Dict[str, Any],  # pass base schema dict in
-        extension_schema: Optional[dict] = None,  # existing extension or None
-    ) -> Any:
-        if (
-            os.getenv("SST_SKIP_EXT_GEN") == "1"
-        ):  # skip using workspace client for tests
-            LOGGER.info("SST_SKIP_EXT_GEN=1; skipping Databricks extension generation.")
-            return None
-
-        inst_name = inst_query.name
-        inst_id = str(inst_query.id)
-
-        mapping = {
-            "course": [
-                "course.csv",
-                "courses.csv",
-                r"^(?=.*AR_DEIDENTIFIED)(?=.*COURSE).*\.csv$",
-            ],
-            "student": ["student.csv", r"^(?=.*AR_DEIDENTIFIED)(?!.*COURSE).*\.csv$"],
-            "semester": ["semester.csv"],
-        }
-
-        key = self.get_key_for_file(mapping, file_name)  # e.g., "student"
-        if key is None:
-            raise HTTPException(
-                404, detail=f"{file_name} not found in {inst_name} validation_mapping"
-            )
-
-        key_lc = key.lower()
-
-        # 4) If this model already exists in the provided extension for this institution, skip
-        if extension_schema is not None:
-            if not isinstance(extension_schema, dict):
-                raise HTTPException(
-                    400, detail="extension_schema must be a dict if provided"
-                )
-
-            inst_block = extension_schema.get("institutions", {}).get(inst_id, {})
-            data_models = inst_block.get("data_models", {})
-            existing_keys_lc = {str(k).lower() for k in data_models.keys()}
-
-            if key_lc in existing_keys_lc:
-                LOGGER.info(
-                    "Model '%s' already present for institution '%s' — skipping (return None).",
-                    key,
-                    inst_id,
-                )
-                return None  # <-- sentinel: do not write
-
-        # 5) Read the unvalidated CSV from GCS
-        try:
-            client = storage.Client()
-            bucket = client.bucket(bucket_name)
-            blob = bucket.blob(f"unvalidated/{file_name}")
-            with blob.open("r") as fh:
-                df = pd.read_csv(fh)
-        except Exception as e:
-            LOGGER.exception("Failed to read %s from GCS", file_name)
-            raise HTTPException(500, detail=f"Failed to read {file_name} from GCS: {e}")
-
-        updated_extension = generate_extension_schema(
-            df=df,
-            models=key,  # exactly one model
-            institution_id=inst_id,
-            base_schema=base_schema,  # reference only, not mutated
-            existing_extension=extension_schema,  # may be None
-        )
-
-        return updated_extension

src/webapp/gcsutil.py

@@ -378,7 +378,7 @@ def validate_file(
             base_schema: Base schema dict.
             inst_schema: Optional extension schema with institutions.* blocks.
             institution_id: Key into inst_schema["institutions"]: "edvise", "pdp",
-                "legacy" (any-format uploads), or institution UUID for custom. Default "pdp".
+                or "legacy" (any-format uploads). Default "pdp".
             institution_identifier: Optional institution ID (e.g. UUID). Reserved for
                 future use; Edvise uses JSON-based validation only (different shape).
 

src/webapp/routers/data.py

@@ -1340,31 +1340,6 @@ def _get_validation_base_schema(sess: Session) -> Tuple[Any, Any, float]:
     return (base_schema_id, base_schema, now)
 
 
-def _ext_models_set(doc: Optional[dict], inst: Any, inst_id: str) -> set[str]:
-    """Extract model keys from extension document (root or institutions.* layout).
-
-    Args:
-        doc: Extension schema JSON doc (or None).
-        inst: Institution row (for id in institutions lookup).
-        inst_id: Institution id string (for institutions lookup).
-
-    Returns:
-        Set of lowercase model names (e.g. {"student", "course"}).
-    """
-    if not doc or not isinstance(doc, dict):
-        return set()
-    if isinstance(doc.get("data_models"), dict):
-        return {str(k).lower() for k in doc["data_models"].keys()}
-    inst_key_candidates = {str(getattr(inst, "id", "")), inst_id}
-    insts = doc.get("institutions", {})
-    if isinstance(insts, dict):
-        for key in inst_key_candidates:
-            block = insts.get(key)
-            if isinstance(block, dict) and isinstance(block.get("data_models"), dict):
-                return {str(k).lower() for k in block["data_models"].keys()}
-    return set()
-
-
 def _resolve_edvise_schema(
     sess: Session, now: float
 ) -> Tuple[str, Optional[Dict[str, Any]]]:
@@ -1422,109 +1397,6 @@ def _resolve_pdp_schema(
     return (schema_namespace, inst_schema)
 
 
-def _persist_custom_schema_extension(
-    sess: Session,
-    inst_id: str,
-    schema_extension: Dict[str, Any],
-    base_schema_id: Any,
-    cache_key: str,
-) -> None:
-    """Deactivate existing extension records and insert new one; update cache."""
-    import time
-
-    existing_extensions = (
-        sess.execute(
-            select(SchemaRegistryTable).where(
-                SchemaRegistryTable.inst_id == str_to_uuid(inst_id),
-                SchemaRegistryTable.doc_type == DocType.extension,
-                SchemaRegistryTable.is_active.is_(True),
-            )
-        )
-        .scalars()
-        .all()
-    )
-    for existing in existing_extensions:
-        existing.is_active = False
-    new_record = SchemaRegistryTable(
-        doc_type=DocType.extension,
-        inst_id=str_to_uuid(inst_id),
-        is_pdp=False,  # type: ignore
-        version_label="1.0.0",
-        extends_schema_id=base_schema_id,
-        json_doc=schema_extension,
-        is_active=True,
-    )
-    sess.add(new_record)
-    sess.flush()
-    logging.info(
-        "Schema record inserted for '%s' (deactivated %d existing)",
-        inst_id,
-        len(existing_extensions),
-    )
-    STATE._ext_cache[cache_key] = (time.monotonic() + EXT_TTL, schema_extension)
-
-
-def _resolve_custom_schema(
-    sess: Session,
-    inst: Any,
-    inst_id: str,
-    now: float,
-    allowed_schemas: List[str],
-    bucket: str,
-    base_schema: dict,
-    base_schema_id: Any,
-    file_name: str,
-) -> Tuple[str, Optional[Dict[str, Any]]]:
-    """Resolve schema namespace and extension for custom (non-PDP/ES/legacy) institutions."""
-    schema_namespace = str(getattr(inst, "id", ""))
-    ext_cache = STATE._ext_cache
-    key = str(getattr(inst, "id", ""))
-    cached = ext_cache.get(key)
-    if cached and now < cached[0]:
-        inst_schema = cached[1]
-    else:
-        inst_schema = sess.execute(
-            select(SchemaRegistryTable.json_doc)
-            .where(
-                SchemaRegistryTable.inst_id == getattr(inst, "id", None),
-                SchemaRegistryTable.is_active.is_(True),
-                SchemaRegistryTable.doc_type == DocType.extension,
-            )
-            .limit(1)
-        ).scalar_one_or_none()
-        ext_cache[key] = (now + EXT_TTL, inst_schema)
-    inferred_lower = {m.lower() for m in allowed_schemas}
-    ext_models = _ext_models_set(inst_schema, inst, inst_id)
-    if inferred_lower.issubset(ext_models):
-        return (schema_namespace, inst_schema)
-    dbc = DatabricksControl()
-    schema_extension: Optional[Dict[str, Any]] = dbc.create_custom_schema_extension(
-        bucket_name=bucket,
-        inst_query=inst,
-        file_name=file_name,
-        base_schema=base_schema,
-        extension_schema=inst_schema,
-    )
-    if schema_extension is not None:
-        try:
-            _persist_custom_schema_extension(
-                sess, inst_id, schema_extension, base_schema_id, key
-            )
-        except IntegrityError as e:
-            sess.rollback()
-            logging.warning("IntegrityError: %s", e)
-        except Exception as e:
-            sess.rollback()
-            logging.error("Unexpected DB error: %s", e)
-            raise HTTPException(
-                status_code=500,
-                detail=f"Unexpected database error while inserting file record: {e}",
-            )
-        return (schema_namespace, schema_extension)
-    logging.info("No-op: extension already contains this model for inst %s", inst_id)
-    return (schema_namespace, inst_schema)
-
-
 def _resolve_schema_namespace_and_extension(
     sess: Session,
     inst: Any,
@@ -1536,7 +1408,7 @@ def _resolve_schema_namespace_and_extension(
     base_schema_id: Any,
     file_name: str,
 ) -> Tuple[str, Optional[Dict[str, Any]]]:
-    """Resolve schema_namespace and updated_inst_schema by institution type (edvise/pdp/legacy/custom)."""
+    """Resolve schema_namespace and updated_inst_schema by institution type (edvise/pdp/legacy)."""
     pdp_id = getattr(inst, "pdp_id", None)
     edvise_id = getattr(inst, "edvise_id", None)
     legacy_id = getattr(inst, "legacy_id", None)
@@ -1552,16 +1424,12 @@ def _resolve_schema_namespace_and_extension(
         return _resolve_pdp_schema(sess, now)
     if legacy_id:
         return ("legacy", None)
-    return _resolve_custom_schema(
-        sess,
-        inst,
-        inst_id,
-        now,
-        allowed_schemas,
-        bucket,
-        base_schema,
-        base_schema_id,
-        file_name,
+    raise HTTPException(
+        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        detail=(
+            "Institution configuration error: institution has no pdp_id, edvise_id, "
+            "or legacy_id; cannot resolve validation schema."
+        ),
     )
 
 

src/webapp/routers/data_test.py

@@ -34,7 +34,6 @@
     DataOverview,
     DataInfo,
     _infer_allowed_schemas_from_filename,
-    _ext_models_set,
 )
 from fastapi import HTTPException
 from ..gcsutil import StorageControl
@@ -164,6 +163,9 @@ def session_fixture():
                     InstTable(
                         id=USER_VALID_INST_UUID,
                         name="school_1",
+                        legacy_id="legacy_test",
+                        pdp_id=None,
+                        edvise_id=None,
                         created_at=DATETIME_TESTING,
                         updated_at=DATETIME_TESTING,
                     ),
@@ -917,7 +919,7 @@ def legacy_session_fixture():
                         legacy_id="legacy123",
                         pdp_id=None,
                         edvise_id=None,
-                        schemas=["STUDENT", "COURSE"],
+                        schemas=["UNKNOWN"],
                         created_at=DATETIME_TESTING,
                         updated_at=DATETIME_TESTING,
                     ),
@@ -1271,36 +1273,6 @@ def test_infer_allowed_schemas_non_legacy_arbitrary_raises_422() -> None:
     assert "random" in exc_info.value.detail
 
 
-def test_ext_models_set_none_returns_empty() -> None:
-    """None doc returns empty set."""
-    inst = _make_inst()
-    assert _ext_models_set(None, inst, "inst-id") == set()
-
-
-def test_ext_models_set_root_data_models() -> None:
-    """Doc with root data_models returns lowercase keys."""
-    inst = _make_inst()
-    doc: dict[str, Any] = {"data_models": {"STUDENT": {}, "COURSE": {}}}
-    assert _ext_models_set(doc, inst, "x") == {"course", "student"}
-
-
-def test_ext_models_set_institutions_block() -> None:
-    """Doc with institutions[inst_id].data_models returns keys."""
-    inst = _make_inst()
-    inst.id = uuid.UUID("12345678-1234-1234-1234-123456789abc")  # type: ignore
-    doc: dict[str, Any] = {
-        "institutions": {
-            "12345678123412341234123456789abc": {
-                "data_models": {"student": {}, "course": {}},
-            }
-        }
-    }
-    assert _ext_models_set(doc, inst, "12345678123412341234123456789abc") == {
-        "course",
-        "student",
-    }
-
-
 def test_validate_edvise_non_descriptive_filename_returns_422(
     edvise_client: TestClient,
 ) -> None:
@@ -1413,6 +1385,35 @@ def test_validation_helper_pdp_and_edvise_mutual_exclusivity(
     edvise_session.commit()
 
 
+def test_validation_helper_rejects_institution_without_school_type(
+    edvise_client: TestClient, edvise_session: sqlalchemy.orm.Session
+) -> None:
+    """Upload validation requires pdp_id, edvise_id, or legacy_id on the institution."""
+    inst = edvise_session.execute(
+        select(InstTable).where(InstTable.id == EDVISE_INST_UUID)
+    ).scalar_one()
+    saved = (inst.edvise_id, inst.pdp_id, inst.legacy_id)
+    inst.edvise_id = None  # type: ignore
+    inst.pdp_id = None  # type: ignore
+    inst.legacy_id = None  # type: ignore
+    edvise_session.commit()
+
+    from .data import STATE
+
+    STATE._edvise_cache = (0.0, None)
+
+    response = edvise_client.post(
+        "/institutions/"
+        + uuid_to_str(EDVISE_INST_UUID)
+        + "/input/validate-upload/test_student_file.csv",
+    )
+    assert response.status_code == 500
+    assert "no pdp_id, edvise_id, or legacy_id" in response.json()["detail"]
+
+    inst.edvise_id, inst.pdp_id, inst.legacy_id = saved
+    edvise_session.commit()
+
+
 def test_edvise_schema_cache(
     edvise_client: TestClient, edvise_session: sqlalchemy.orm.Session
 ) -> None: