Skip to content

cdm_data_loaders.readers.jsonschema_xsv.xsv_validator.schema_utils

XSV-specific JSON Schema validation and header generation utilities.

ValidatedXsvSchema pydantic-model

Bases: ValidatedSchema

A JSON Schema to be used for parsing and validating XSV data.

Show JSON schema: 
{
  "description": "A JSON Schema to be used for parsing and validating XSV data.",
  "properties": {
    "jsonschema": {
      "additionalProperties": true,
      "title": "Jsonschema",
      "type": "object"
    }
  },
  "required": [
    "jsonschema"
  ],
  "title": "ValidatedXsvSchema",
  "type": "object"
}

Fields:

  • jsonschema (dict[str, Any])

Validators:

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/schema_utils.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class ValidatedXsvSchema(ValidatedSchema):
    """A JSON Schema to be used for parsing and validating XSV data."""

    @computed_field
    @property
    def has_xsv_parser_config(self) -> bool:
        """Whether or not this schema has custom fields for parsing XSV files."""
        return len(self.jsonschema.get(X_XSV_CONFIG_KEY, {}).keys()) > 0

    @computed_field
    @property
    def has_xsv_metaschema(self) -> bool:
        """Whether or not this schema has the custom metaschema URI indicating it holds XSV metadata."""
        return self.jsonschema.get("$schema") == CUSTOM_META_SCHEMA_URI

    @computed_field
    @property
    def required_cols(self) -> list[str]:
        """Retrieve the list of headers from the top-level ``required`` field."""
        return self.jsonschema.get("required", [])

    @field_validator("jsonschema", mode="after")
    @classmethod
    def validate_jsonschema(cls, schema: dict[str, Any]) -> dict[str, Any]:
        """Validate the JSON schema in the jsonschema field and ensure it can be used for xsv validation.

        :param schema: the schema
        :type  schema: dict[str, Any]
        :raises ValueError: if there is no ``$schema`` field to identify the metaschema used
        :raises jsonschema.exceptions.SchemaError: if the schema is invalid
        :raises jsonschema.exceptions.ValidationError: if the schema fails validation
        :raises ValueError: if there are no ``required`` fields to indicate the CSV columns and order
        :return: the validated schema
        :rtype : dict[str, Any]
        """
        validated_schema = super().validate_jsonschema(schema)
        required_cols = validated_schema.get("required")
        if not isinstance(required_cols, list) or not required_cols:
            err_msg = "Could not find any required cols in schema"
            raise ValueError(err_msg)
        return validated_schema

has_xsv_parser_config property 

has_xsv_parser_config: bool

Whether or not this schema has custom fields for parsing XSV files.

has_xsv_metaschema property 

has_xsv_metaschema: bool

Whether or not this schema has the custom metaschema URI indicating it holds XSV metadata.

required_cols property 

required_cols: list[str]

Retrieve the list of headers from the top-level required field.

validate_jsonschema pydantic-validator 

validate_jsonschema(
    schema: dict[str, Any],
) -> dict[str, Any]

Validate the JSON schema in the jsonschema field and ensure it can be used for xsv validation.

Parameters:

Name Type Description Default
schema dict[str, Any]

the schema

 required

Returns:

Type Description
dict[str, Any]

the validated schema

Raises:

Type Description
ValueError

if there is no $schema field to identify the metaschema used
jsonschema.exceptions.SchemaError

if the schema is invalid
jsonschema.exceptions.ValidationError

if the schema fails validation
ValueError

if there are no required fields to indicate the CSV columns and order
Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/schema_utils.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@field_validator("jsonschema", mode="after")
@classmethod
def validate_jsonschema(cls, schema: dict[str, Any]) -> dict[str, Any]:
    """Validate the JSON schema in the jsonschema field and ensure it can be used for xsv validation.

    :param schema: the schema
    :type  schema: dict[str, Any]
    :raises ValueError: if there is no ``$schema`` field to identify the metaschema used
    :raises jsonschema.exceptions.SchemaError: if the schema is invalid
    :raises jsonschema.exceptions.ValidationError: if the schema fails validation
    :raises ValueError: if there are no ``required`` fields to indicate the CSV columns and order
    :return: the validated schema
    :rtype : dict[str, Any]
    """
    validated_schema = super().validate_jsonschema(schema)
    required_cols = validated_schema.get("required")
    if not isinstance(required_cols, list) or not required_cols:
        err_msg = "Could not find any required cols in schema"
        raise ValueError(err_msg)
    return validated_schema

validate_jsonschema 

validate_jsonschema(
    schema_path: Path,
) -> ValidatedXsvSchema

Ensure that a given file contains a valid XSV JSON schema.

Parameters:

Name Type Description Default
schema_path Path

path to the JSON Schema file

 required

Returns:

Type Description
ValidatedXsvSchema

validated XSV JSON Schema

Raises:

Type Description
jsonschema.exceptions.SchemaError

if the schema is invalid
Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/schema_utils.py 
67
68
69
70
71
72
73
74
75
76
77
def validate_jsonschema(schema_path: Path) -> ValidatedXsvSchema:
    """Ensure that a given file contains a valid XSV JSON schema.

    :param schema_path: path to the JSON Schema file
    :type schema_path: Path
    :raises jsonschema.exceptions.SchemaError: if the schema is invalid
    :return: validated XSV JSON Schema
    :rtype: ValidatedXsvSchema
    """
    schema = json.loads(schema_path.read_bytes())
    return ValidatedXsvSchema(jsonschema=schema)

generate_first_pass_schema 

generate_first_pass_schema(
    validated_schema: ValidatedXsvSchema,
) -> ValidatedXsvSchema

Given a full schema file, generate a schema to perform a loose first-pass validation with.

The first pass schema is used for verifying that the top-level columns are correct; no further checks are performed.

Parameters:

Name Type Description Default
validated_schema ValidatedXsvSchema

parsed, validated JSON schema

 required

Returns:

Type Description
ValidatedXsvSchema

the validated first pass schema
Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/schema_utils.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@validate_call
def generate_first_pass_schema(validated_schema: ValidatedXsvSchema) -> ValidatedXsvSchema:
    """Given a full schema file, generate a schema to perform a loose first-pass validation with.

    The first pass schema is used for verifying that the top-level columns are correct; no further checks
    are performed.

    :param validated_schema: parsed, validated JSON schema
    :type validated_schema: ValidatedXsvSchema
    :return: the validated first pass schema
    :rtype: ValidatedXsvSchema
    """
    first_pass_schema = {
        k: v for k, v in validated_schema.jsonschema.items() if k in ["$schema", "$id", "title", "required"]
    }

    first_pass_schema["type"] = "object"
    first_pass_schema["properties"] = {req: {"type": ["string", "null"]} for req in validated_schema.required_cols}

    validator = jsonschema.validators.validator_for(first_pass_schema)
    validator.check_schema(first_pass_schema)

    return ValidatedXsvSchema(jsonschema=first_pass_schema)

generate_header 

generate_header(
    validated_schema: ValidatedXsvSchema,
    target_dir: DirectoryPath,
    header_file_name: NonEmptyStr = "header.txt",
    delimiter: CharStr = "\t",
) -> Path

Generate a header file for an xSV file from a JSON Schema.

If no file name is supplied, the header file is saved as header.txt.

The top level required properties are assumed to be the columns of the xSV file.

Parameters:

Name Type Description Default
validated_schema ValidatedXsvSchema

parsed, validated JSON schema

 required
target_dir DirectoryPath

directory in which to save the file

 required
header_file_name NonEmptyStr

name for the header file, defaults to "header.txt"

 'header.txt'
delimiter CharStr

delimiter to use for the headers, defaults to "\t"

 '\t'

Returns:

Type Description
Path

path to the newly-created header.txt file
Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/schema_utils.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@validate_call
def generate_header(
    validated_schema: ValidatedXsvSchema,
    target_dir: DirectoryPath,
    header_file_name: NonEmptyStr = "header.txt",
    delimiter: CharStr = "\t",
) -> Path:
    r"""Generate a header file for an xSV file from a JSON Schema.

    If no file name is supplied, the header file is saved as ``header.txt``.

    The top level required properties are assumed to be the columns of the xSV file.

    :param validated_schema: parsed, validated JSON schema
    :type validated_schema: ValidatedXsvSchema
    :param target_dir: directory in which to save the file
    :type target_dir: DirectoryPath
    :param header_file_name: name for the header file, defaults to "header.txt"
    :type header_file_name: NonEmptyStr, optional
    :param delimiter: delimiter to use for the headers, defaults to "\t"
    :type delimiter: CharStr, optional
    :return: path to the newly-created header.txt file
    :rtype: Path
    """
    header_file_path = target_dir / header_file_name
    header_file_path.write_text(delimiter.join(validated_schema.required_cols) + "\n")
    return header_file_path

get_schema_parsing_metadata 

get_schema_parsing_metadata(
    validated_schema: ValidatedXsvSchema,
) -> dict[str, Any]

Read in the schema parser config information from the supplied schema.

Parameters:

Name Type Description Default
validated_schema ValidatedXsvSchema

parsed, validated JSON schema

 required

Returns:

Type Description
dict[str, Any]

parsed xsv schema metadata

Raises:

Type Description
ValueError

if no xsv config information is found in the schema
Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/schema_utils.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def get_schema_parsing_metadata(validated_schema: ValidatedXsvSchema) -> dict[str, Any]:
    """Read in the schema parser config information from the supplied schema.

    :param validated_schema: parsed, validated JSON schema
    :type validated_schema: ValidatedXsvSchema
    :raises ValueError: if no xsv config information is found in the schema
    :return: parsed xsv schema metadata
    :rtype: dict[str, Any]
    """
    if not validated_schema.has_xsv_parser_config:
        err_msg = "No xsv config information found in schema"
        raise ValueError(err_msg)

    return {
        k.replace("x-", "").replace("-", "_"): v
        for k, v in validated_schema.jsonschema.get(X_XSV_CONFIG_KEY, {}).items()
    }