diff --git a/CHANGELOG.md b/CHANGELOG.md index abe4fb2e..6a587be2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,6 @@ +## 1.7.0 (2026-09-22) +- feat: validate PDP uploads with bronze `dataio` converters (#306) + ## 1.6.0 (2026-09-16) - feat: return eligible academic terms via `/{inst_id}/eligible-inference-terms` and pass `term_filter` to inference (#201) - feat: add `archived_at` timestamp to models (#300) diff --git a/pyproject.toml b/pyproject.toml index 97ed4109..5353d5b8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "src" -version = "1.6.0" +version = "1.7.0" description = "School-agnostic lib for implementing Edvise workflows." readme = "README.md" requires-python = ">=3.12,<3.13" diff --git a/src/webapp/gcsutil.py b/src/webapp/gcsutil.py index a289a1e1..32e81cb8 100644 --- a/src/webapp/gcsutil.py +++ b/src/webapp/gcsutil.py @@ -396,8 +396,9 @@ def validate_file( file_name: Blob name under unvalidated/. allowed_schemas: List of schema/model names allowed. institution_id: Validation namespace: "edvise", "pdp", or "legacy". - institution_identifier: For ES, institution name used to fetch bronze - ``training_inputs/dataio.py`` converters. Unused for PDP/Legacy. + institution_identifier: Institution name used to fetch bronze + ``training_inputs/dataio.py`` converters for PDP and ES. Unused + for Legacy. Returns: List of inferred schema names (e.g. ["STUDENT"]). diff --git a/src/webapp/routers/data.py b/src/webapp/routers/data.py index 58ee00e3..24c61cc9 100644 --- a/src/webapp/routers/data.py +++ b/src/webapp/routers/data.py @@ -1979,7 +1979,7 @@ def _run_validation_and_upsert_file_record( allowed_schemas, institution_id=schema_namespace, institution_identifier=( - institution_name if schema_namespace == "edvise" else None + institution_name if schema_namespace in ("edvise", "pdp") else None ), ) except HardValidationError as e: diff --git a/src/webapp/routers/data_test.py b/src/webapp/routers/data_test.py index 41ce872d..34d36709 100644 --- a/src/webapp/routers/data_test.py +++ b/src/webapp/routers/data_test.py @@ -2143,6 +2143,9 @@ def test_validate_upload_pdp_only_institution_skips_bronze_sync( ) assert response.status_code == 200 + call_kwargs = MOCK_STORAGE.validate_file.call_args.kwargs + assert call_kwargs.get("institution_id") == "pdp" + assert call_kwargs.get("institution_identifier") == "pdp_only_school" MOCK_DATABRICKS.run_validated_gcs_to_bronze_sync.assert_not_called() diff --git a/src/webapp/validation.py b/src/webapp/validation.py index 7ce42e38..8c586afe 100644 --- a/src/webapp/validation.py +++ b/src/webapp/validation.py @@ -1,11 +1,10 @@ """File validation functions for upload workflows. -PDP uploads validate through Pandera schemas imported from the ``edvise`` -package (optional school converters may be passed in). Edvise Schema (ES) -uploads fetch school ``training_inputs/dataio.py`` converters and -``training_inputs/config.toml`` grade maps from the institution bronze volume -(when present), then validate via ``read_raw_es_*`` + ES Pandera schemas — -parity with ES Databricks data-audit jobs. Legacy uploads use any-format CSV +PDP and Edvise Schema (ES) uploads fetch school ``training_inputs/dataio.py`` +converters from the institution bronze volume when present (soft-fallback to +defaults). ES also loads ``training_inputs/config.toml`` grade maps, then +validates via ``read_raw_es_*`` + ES Pandera schemas. PDP validates via +``read_raw_pdp_*`` + PDP Pandera schemas. Legacy uploads use any-format CSV read plus a PII column-name guard. The old API-local JSON schema validation path has been removed. """ @@ -90,10 +89,11 @@ def validate_file_reader( filename: Path or file-like object for the CSV. allowed_schema: List of model names to validate against. institution_id: Validation namespace: "edvise", "pdp", or "legacy". - institution_identifier: For ES, institution name for bronze ``dataio`` lookup. - pdp_cohort_converter_func: Optional cohort row transform before Pandera; default - None. Batch PDP jobs may still apply school-specific cohort converters via ``dataio``. - pdp_course_converter_func: Optional course converter; default duplicate handling only. + institution_identifier: Institution name for bronze ``dataio`` lookup (PDP and ES). + pdp_cohort_converter_func: Optional cohort row transform before Pandera. When + omitted, PDP loads ``converter_func_cohort`` from bronze ``dataio.py``. + pdp_course_converter_func: Optional course converter. When omitted, PDP loads + ``converter_func_course`` from bronze, else default duplicate handling. Returns: Dict with validation_status, schemas, missing_optional, unknown_extra_columns, @@ -228,9 +228,9 @@ def _model_list_from_models(models: Union[str, List[str], None]) -> List[str]: # --------------------------------------------------------------------------- # -# PDP single-model path: edvise read + Pandera validate. Cohort converter defaults -# to None so PDP validated row sets can differ from batch jobs that use dataio -# converters. +# PDP single-model path: bronze dataio converters (when present) + edvise read +# + Pandera validate. Soft-falls back to no cohort converter / default course +# duplicate handling when bronze dataio is missing — same as data-audit jobs. # --------------------------------------------------------------------------- # # Datetime formats for ES cohort/course (same order as es_data_audit) @@ -278,7 +278,7 @@ def _import_dataio_module_isolated(inst_name: str, file_path: str) -> Any: converters from different institutions never collide in ``sys.modules``. """ safe = re.sub(r"[^a-zA-Z0-9_]", "_", inst_name) or "unknown" - module_name = f"es_bronze_dataio_{safe}_{uuid.uuid4().hex}" + module_name = f"bronze_dataio_{safe}_{uuid.uuid4().hex}" spec = importlib.util.spec_from_file_location(module_name, file_path) if spec is None or spec.loader is None: raise ImportError(f"Could not create import spec for {file_path}") @@ -301,7 +301,7 @@ def load_es_converters_from_bronze( load ``converter_func_cohort`` / ``converter_func_course`` when present. Soft-falls back to ``(None, None)`` on missing file, download failure, import - failure, or missing attributes — parity with ES Databricks data-audit jobs. + failure, or missing attributes — parity with PDP/ES Databricks data-audit jobs. Fresh fetch per call; no process-wide shared ``dataio`` module. Converter *runtime* errors during validation are not soft-fallbacked; they @@ -322,7 +322,7 @@ def load_es_converters_from_bronze( inst_name, relative_path="dataio.py" ) raw = _read_stream_to_bytes(stream) - fd, tmp_path = tempfile.mkstemp(suffix="_dataio.py", prefix="es_bronze_") + fd, tmp_path = tempfile.mkstemp(suffix="_dataio.py", prefix="bronze_") try: os.write(fd, raw) finally: @@ -331,7 +331,7 @@ def load_es_converters_from_bronze( module = _import_dataio_module_isolated(inst_name, tmp_path) except Exception as e: logger.warning( - "ES bronze dataio unavailable for institution=%s; validating without " + "Bronze dataio unavailable for institution=%s; validating without " "converters: %s", inst_name, e, @@ -348,22 +348,22 @@ def load_es_converters_from_bronze( course_converter: PDPConverterFunc = None try: cohort_converter = module.converter_func_cohort - logger.info("Loaded custom ES cohort converter for institution=%s", inst_name) + logger.info("Loaded custom cohort converter for institution=%s", inst_name) except Exception as e: logger.info( - "Running ES validation with default cohort converter for institution=%s", + "Running validation with default cohort converter for institution=%s", inst_name, ) - logger.warning("Failed to load custom ES cohort converter: %s", e) + logger.warning("Failed to load custom cohort converter: %s", e) try: course_converter = module.converter_func_course - logger.info("Loaded custom ES course converter for institution=%s", inst_name) + logger.info("Loaded custom course converter for institution=%s", inst_name) except Exception as e: logger.info( - "Running ES validation with default course converter for institution=%s", + "Running validation with default course converter for institution=%s", inst_name, ) - logger.warning("Failed to load custom ES course converter: %s", e) + logger.warning("Failed to load custom course converter: %s", e) if cohort_converter is not None and not callable(cohort_converter): logger.warning( @@ -777,10 +777,8 @@ def _validate_pdp_with_edvise_read( Validate a single-model PDP cohort or course file via edvise read and Pandera. Writes file-like inputs to a temp path, then calls ``read_raw_pdp_cohort_data`` - (STUDENT) or ``_read_pdp_course_edvise`` (COURSE). Cohort rows are only - transformed when ``pdp_cohort_converter_func`` is set; batch jobs may still - filter cohort rows via ``dataio``, so API output rows are not guaranteed to - match pipeline output for the same file. + (STUDENT) or ``_read_pdp_course_edvise`` (COURSE). School bronze ``dataio`` + converters are applied when passed in by ``validate_dataset``. Args: filename: Path or file-like CSV source. @@ -925,8 +923,9 @@ def validate_dataset( Validate a dataset using the active institution upload workflow. Detects encoding, then routes to Legacy any-format handling, ES - ``read_raw_es_*`` + Pandera (with optional bronze ``dataio`` converters), or - PDP repo Pandera validation for supported single-model STUDENT/COURSE uploads. + ``read_raw_es_*`` + Pandera, or PDP ``read_raw_pdp_*`` + Pandera. PDP and ES + load optional bronze ``dataio`` converters when ``institution_identifier`` + is set. Other model sets are rejected explicitly; the API-local JSON schema validation fallback has been removed. @@ -935,11 +934,12 @@ def validate_dataset( models: Model name(s) to validate. institution_id: Validation namespace (``"pdp"``, ``"edvise"``, or ``"legacy"``). ``"legacy"`` skips Pandera; ``"edvise"`` and ``"pdp"`` use repo schemas. - institution_identifier: For ES, the institution name used to resolve the - bronze volume ``training_inputs/dataio.py`` path. Unused for PDP/Legacy. - pdp_cohort_converter_func: Optional cohort transform before Pandera; default ``None``. - Batch PDP jobs may still apply school-specific cohort converters via ``dataio``. - pdp_course_converter_func: Optional course converter before default duplicate handling. + institution_identifier: Institution name used to resolve bronze + ``training_inputs/dataio.py`` for PDP and ES. Unused for Legacy. + pdp_cohort_converter_func: Optional cohort transform before Pandera. When + omitted, loaded from bronze ``dataio`` for PDP uploads. + pdp_course_converter_func: Optional course converter. When omitted, loaded + from bronze ``dataio`` for PDP uploads (else default duplicate handling). Returns: Dict with validation_status, schemas, missing_optional, unknown_extra_columns, @@ -1003,6 +1003,23 @@ def validate_dataset( schema_class = pdp_edvise.get_edvise_schema_for_upload(institution_id, model_list) if schema_class is not None: + # PDP: same bronze dataio fetch as ES (soft-fallback). Explicit converter + # kwargs still win so tests and callers can override. + if institution_identifier and ( + pdp_cohort_converter_func is None or pdp_course_converter_func is None + ): + bronze_cohort, bronze_course = load_es_converters_from_bronze( + institution_identifier + ) + if pdp_cohort_converter_func is None: + pdp_cohort_converter_func = bronze_cohort + if pdp_course_converter_func is None: + pdp_course_converter_func = bronze_course + elif not institution_identifier: + logger.warning( + "PDP validation without institution_identifier; validating without " + "bronze dataio converters" + ) return _validate_pdp_with_edvise_read( filename, enc, diff --git a/src/webapp/validation_es_test.py b/src/webapp/validation_es_test.py index 61df2912..be618f68 100644 --- a/src/webapp/validation_es_test.py +++ b/src/webapp/validation_es_test.py @@ -102,9 +102,7 @@ def download_side_effect(inst_name: str, relative_path: str = "dataio.py") -> An assert cohort_a(df).attrs["school_marker"] == "inst_alpha" assert cohort_b(df).attrs["school_marker"] == "inst_beta" - bronze_modules = [ - name for name in sys.modules if name.startswith("es_bronze_dataio_") - ] + bronze_modules = [name for name in sys.modules if name.startswith("bronze_dataio_")] assert len(bronze_modules) >= 2 @@ -116,8 +114,8 @@ def test_import_dataio_module_isolated_unique_names(tmp_path: Path) -> None: mod2 = _import_dataio_module_isolated("school_two", str(path)) assert mod1 is not mod2 assert mod1.__name__ != mod2.__name__ - assert mod1.__name__.startswith("es_bronze_dataio_school_one_") - assert mod2.__name__.startswith("es_bronze_dataio_school_two_") + assert mod1.__name__.startswith("bronze_dataio_school_one_") + assert mod2.__name__.startswith("bronze_dataio_school_two_") assert "dataio" not in sys.modules @@ -263,8 +261,10 @@ def fake_read(**kwargs: Any) -> pd.DataFrame: assert "converter blew up" in str(exc_info.value) -def test_pdp_routing_unchanged_does_not_load_es_converters(tmp_path: Path) -> None: - """PDP uploads do not fetch bronze ES dataio converters.""" +def test_pdp_without_institution_identifier_does_not_load_bronze_converters( + tmp_path: Path, +) -> None: + """PDP uploads skip bronze dataio when institution_identifier is omitted.""" csv_path = tmp_path / "cohort.csv" pd.DataFrame({"x": [1]}).to_csv(csv_path, index=False) @@ -297,6 +297,52 @@ def test_pdp_routing_unchanged_does_not_load_es_converters(tmp_path: Path) -> No mock_grade.assert_not_called() +def test_pdp_loads_bronze_converters_when_institution_identifier_set( + tmp_path: Path, +) -> None: + """PDP uploads fetch bronze dataio converters the same way ES does.""" + csv_path = tmp_path / "cohort.csv" + pd.DataFrame({"x": [1]}).to_csv(csv_path, index=False) + + def cohort_converter(df: pd.DataFrame) -> pd.DataFrame: + return df + + def course_converter(df: pd.DataFrame) -> pd.DataFrame: + return df + + with ( + patch( + "src.webapp.validation.load_es_converters_from_bronze", + return_value=(cohort_converter, course_converter), + ) as mock_load, + patch( + "src.webapp.validation.load_es_institution_grade_map_from_bronze", + ) as mock_grade, + patch( + "src.webapp.validation._validate_pdp_with_edvise_read", + return_value={ + "validation_status": "passed", + "schemas": ["STUDENT"], + "missing_optional": [], + "unknown_extra_columns": [], + "normalized_df": pd.DataFrame({"student_id": ["s1"]}), + }, + ) as mock_pdp, + ): + result = validate_file_reader( + str(csv_path), + ["STUDENT"], + institution_id="pdp", + institution_identifier="pdp_school", + ) + + assert result["validation_status"] == "passed" + mock_load.assert_called_once_with("pdp_school") + mock_grade.assert_not_called() + assert mock_pdp.call_args.kwargs["pdp_cohort_converter_func"] is cohort_converter + assert mock_pdp.call_args.kwargs["pdp_course_converter_func"] is course_converter + + def test_chain_es_course_converters_applies_grade_map_before_dataio() -> None: """Job parity: grade_map runs first, then school course converter.""" order: list[str] = [] diff --git a/src/webapp/validation_pdp_edvise.py b/src/webapp/validation_pdp_edvise.py index ad79c49d..c03b743b 100644 --- a/src/webapp/validation_pdp_edvise.py +++ b/src/webapp/validation_pdp_edvise.py @@ -1,7 +1,7 @@ """Pandera schemas re-exported from edvise for upload validation. Imports raw PDP and Edvise schema classes. PDP and ES uploads use these at -upload time (ES after optional bronze ``dataio`` converters). Legacy uploads +upload time after optional bronze ``dataio`` converters. Legacy uploads skip Pandera (any-format CSV + PII guard). This module supplies schema classes and helpers for the PDP/ES validation paths. diff --git a/src/webapp/validation_pdp_read_path_test.py b/src/webapp/validation_pdp_read_path_test.py index 9986fa51..ea2bc4f4 100644 --- a/src/webapp/validation_pdp_read_path_test.py +++ b/src/webapp/validation_pdp_read_path_test.py @@ -65,6 +65,44 @@ def test_validate_file_reader_pdp_student_calls_edvise_read_path( assert call_args[3] == "pdp" +def test_validate_file_reader_pdp_passes_bronze_cohort_converter( + tmp_path: Path, +) -> None: + """PDP STUDENT uploads pass bronze dataio cohort converter into the read path.""" + csv_path = tmp_path / "cohort.csv" + pd.DataFrame({"x": [1]}).to_csv(csv_path, index=False) + + def cohort_converter(df: pd.DataFrame) -> pd.DataFrame: + return df + + with ( + patch( + "src.webapp.validation.load_es_converters_from_bronze", + return_value=(cohort_converter, None), + ) as mock_load, + patch( + "src.webapp.validation._validate_pdp_with_edvise_read", + return_value={ + "validation_status": "passed", + "schemas": ["STUDENT"], + "missing_optional": [], + "unknown_extra_columns": [], + "normalized_df": pd.DataFrame({"student_id": ["s1"]}), + }, + ) as mock_pdp, + ): + result = validate_file_reader( + str(csv_path), + ["STUDENT"], + institution_id="pdp", + institution_identifier="pdp_school", + ) + + assert result["validation_status"] == "passed" + mock_load.assert_called_once_with("pdp_school") + assert mock_pdp.call_args.kwargs["pdp_cohort_converter_func"] is cohort_converter + + def test_validate_file_reader_pdp_course_calls_edvise_read_path(tmp_path: Path) -> None: """When institution_id is pdp and allowed_schema is [COURSE], PDP edvise-read path is used.""" csv_path = tmp_path / "course.csv" @@ -351,7 +389,7 @@ def test_validate_pdp_with_edvise_read_accepts_file_like() -> None: # Edvise read was given a path (temp file when file-like); keyword is file_path assert "file_path" in mock_read.call_args[1] assert isinstance(mock_read.call_args[1]["file_path"], str) - # Cohort validation uses no converter unless pdp_cohort_converter_func is passed + # Cohort validation uses no converter unless pdp_cohort_converter_func or bronze dataio is used assert mock_read.call_args[1]["converter_func"] is None