diff --git a/src/dandisets_linkml_status_tools/cli/models.py b/src/dandisets_linkml_status_tools/cli/models.py index 69e68eb..c2d68fa 100644 --- a/src/dandisets_linkml_status_tools/cli/models.py +++ b/src/dandisets_linkml_status_tools/cli/models.py @@ -1,46 +1,115 @@ +from collections.abc import Sequence from datetime import datetime -from typing import Annotated, Any +from typing import Annotated, Any, Union from dandi.dandiapi import VersionStatus +from jsonschema.exceptions import ValidationError from linkml.validator.report import ValidationResult -from pydantic import BaseModel, Json, PlainSerializer, TypeAdapter +from pydantic import AfterValidator, BaseModel, Json, PlainSerializer, TypeAdapter from typing_extensions import TypedDict # Required for Python < 3.12 by Pydantic -# A `TypedDict` that has a key corresponding to each field in `ValidationResult` -# except for the `instance` field -TrimmedValidationResult = TypedDict( - "TrimmedValidationResult", - { - name: info.annotation - for name, info in ValidationResult.model_fields.items() - if name != "instance" - }, + +class JsonValidationErrorView(BaseModel): + """ + A Pydantic model to represent a `jsonschema.exceptions.ValidationError` object, + by including selective fields or properties of the original object, + for serialization + """ + + message: str + absolute_path: Sequence[Union[str, int]] + absolute_schema_path: Sequence[Union[str, int]] + validator: str + validator_value: Any + + +# Build a `TypedDict` for representing a polished version of `ValidationResult` +field_annotations = { + name: info.annotation + for name, info in ValidationResult.model_fields.items() + if name not in {"instance", "source"} +} +field_annotations["source"] = JsonValidationErrorView +PolishedValidationResult = TypedDict( + "PolishedValidationResult", + field_annotations, ) -def trim_validation_results( - errs: list[ValidationResult], -) -> list[TrimmedValidationResult]: +def check_source_jsonschema_validation_error( + results: list[ValidationResult], +) -> list[ValidationResult]: + """ + Check if the `source` field of each `ValidationResult` object in a given list is a + `jsonschema.exceptions.ValidationError` object. + + :param results: The list of `ValidationResult` objects to be checked. + + :return: The list of `ValidationResult` objects if all `source` fields are + `jsonschema.exceptions.ValidationError` objects. + + :raises ValueError: If the `source` field of a `ValidationResult` object is not a + `jsonschema.exceptions.ValidationError` object. + """ + for result in results: + result_source = result.source + if not isinstance(result_source, ValidationError): + msg = ( + f"Expected `source` field of a `ValidationResult` object to be " + f"a {ValidationError!r} object, but got {result_source!r}" + ) + raise ValueError(msg) # noqa: TRY004 + return results + + +def polish_validation_results( + results: list[ValidationResult], +) -> list[PolishedValidationResult]: """ - Trim the `ValidationResult` objects in a list to exclude their `instance` field. + Polish the `ValidationResult` objects in a list to exclude their `instance` field + and include their `source` field for serialization. - :param errs: The list of `ValidationResult` objects to be trimmed. + Note: This function is intended to be used to handle `ValidationResult` objects + produced by `linkml.validator.plugins.JsonschemaValidationPlugin`. The `source` + field of these `ValidationResult` objects is expected to be a + `jsonschema.exceptions.ValidationError` object. - :return: The list of `TrimmedValidationResult` objects representing the trimmed + :param results: The list of `ValidationResult` objects to be polished. + + :return: The list of `PolishedValidationResult` objects representing the polished `ValidationResult` objects. + + :raises ValueError: If the `source` field of a `ValidationResult` object is not a + `jsonschema.exceptions.ValidationError` object. """ - trimmed_errs = [] - for err in errs: - err_as_dict = err.model_dump() - del err_as_dict["instance"] - trimmed_errs.append(err_as_dict) - return trimmed_errs + polished_results = [] + for result in results: + result_as_dict = result.model_dump() + + # Remove the `instance` field + del result_as_dict["instance"] + + # Include the `source` field as a `JsonValidationErrorView` object + result_source = result.source + # noinspection PyTypeChecker + result_as_dict["source"] = JsonValidationErrorView( + message=result_source.message, + absolute_path=result_source.absolute_path, + absolute_schema_path=result_source.absolute_schema_path, + validator=result_source.validator, + validator_value=result_source.validator_value, + ) + + polished_results.append(result_as_dict) + return polished_results DandisetMetadataType = dict[str, Any] PydanticValidationErrsType = list[dict[str, Any]] LinkmlValidationErrsType = Annotated[ - list[ValidationResult], PlainSerializer(trim_validation_results) + list[ValidationResult], + AfterValidator(check_source_jsonschema_validation_error), + PlainSerializer(polish_validation_results), ] dandiset_metadata_adapter = TypeAdapter(DandisetMetadataType)