Skip to content

cdm_data_loaders.converters.core.io

Schema input parsing and shared converter facade base class.

JsonSchemaConverterBase pydantic-model

Bases: BaseModel

Shared base class for schema converter facades supporting string and file input.

Fields:

Source code in src/cdm_data_loaders/converters/core/io.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class JsonSchemaConverterBase[OutputT](BaseModel):
    """Shared base class for schema converter facades supporting string and file input."""

    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)

    _allow_yaml_in_string: ClassVar[bool] = False
    _run_root_guards: ClassVar[bool] = True
    _error_cls: ClassVar[type[ConversionError]] = ConversionError
    _invalid_schema_error_cls: ClassVar[type[ConversionError]] = ConversionError
    _converter_name: ClassVar[str] = ""
    _root_target_label: ClassVar[str] = ""
    _dereference_function: ClassVar[str] = "cdm_data_loaders.schemas.jsonschema.dereference_schema"

    extension_registry: InstanceOf[ExtensionRegistry] = DEFAULT_EXTENSIONS

    def _reader(self) -> JsonSchemaReader:
        """Configure reader diagnostics for this public facade."""
        return JsonSchemaReader(
            self.extension_registry,
            error_type=self._error_cls,
            converter_name=self._converter_name or type(self).__name__,
            dereference_function=self._dereference_function,
        )

    def _emitter(self) -> Emitter[OutputT]:
        """Return the configured emitter for this facade."""
        raise NotImplementedError

    def _read(self, schema: dict[str, Any]) -> SchemaDocument:
        """Preserve facade guards and error identity around the reader."""
        if self._run_root_guards:
            converter_name = self._converter_name or type(self).__name__
            require_schema_keyword(schema, self._invalid_schema_error_cls, converter_name=converter_name)
            require_object_root(schema, self._error_cls, target=self._root_target_label)
        try:
            return self._reader().read(schema)
        except ConversionError as error:
            if isinstance(error, self._error_cls):
                raise
            raise self._error_cls(str(error)) from error

    def convert(self, schema: dict[str, Any]) -> OutputT:
        """Convert a parsed schema mapping to the target representation."""
        try:
            return self._emitter().emit(self._read(schema))
        except ConversionError as error:
            if isinstance(error, self._error_cls):
                raise
            raise self._error_cls(str(error)) from error

    def convert_from_string(self, schema_str: str) -> OutputT:
        """Convert a schema string according to the converter's text parsing policy."""
        return self.convert(load_schema_text(schema_str, allow_yaml=self._allow_yaml_in_string))

    def convert_from_file(self, path: str) -> OutputT:
        """Convert a JSON or YAML schema file selected by its extension."""
        return self.convert(load_schema_file(path))

convert 

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

Convert a parsed schema mapping to the target representation.

Source code in src/cdm_data_loaders/converters/core/io.py 
104
105
106
107
108
109
110
111
def convert(self, schema: dict[str, Any]) -> OutputT:
    """Convert a parsed schema mapping to the target representation."""
    try:
        return self._emitter().emit(self._read(schema))
    except ConversionError as error:
        if isinstance(error, self._error_cls):
            raise
        raise self._error_cls(str(error)) from error

convert_from_string 

convert_from_string(schema_str: str) -> OutputT

Convert a schema string according to the converter's text parsing policy.

Source code in src/cdm_data_loaders/converters/core/io.py 
113
114
115
def convert_from_string(self, schema_str: str) -> OutputT:
    """Convert a schema string according to the converter's text parsing policy."""
    return self.convert(load_schema_text(schema_str, allow_yaml=self._allow_yaml_in_string))

convert_from_file 

convert_from_file(path: str) -> OutputT

Convert a JSON or YAML schema file selected by its extension.

Source code in src/cdm_data_loaders/converters/core/io.py 
117
118
119
def convert_from_file(self, path: str) -> OutputT:
    """Convert a JSON or YAML schema file selected by its extension."""
    return self.convert(load_schema_file(path))

load_schema_text 

load_schema_text(
    text: str | bytes, *, allow_yaml: bool = False
) -> dict[str, Any]

Parse JSON text, falling back to YAML only when explicitly enabled.

Source code in src/cdm_data_loaders/converters/core/io.py 
25
26
27
28
29
30
31
32
33
def load_schema_text(text: str | bytes, *, allow_yaml: bool = False) -> dict[str, Any]:
    """Parse JSON text, falling back to YAML only when explicitly enabled."""
    try:
        document = json.loads(text)
    except json.JSONDecodeError:
        if not allow_yaml:
            raise
        document = yaml.safe_load(text)
    return _require_mapping(document, "text")

load_schema_file 

load_schema_file(path: str | Path) -> dict[str, Any]

Load a JSON or YAML schema from a file, choosing by the .json extension.

Parameters:

Name Type Description Default
path str | Path

path to the schema file

 required

Returns:

Type Description
dict[str, Any]

the parsed schema document
Source code in src/cdm_data_loaders/converters/core/io.py 
36
37
38
39
40
41
42
43
44
45
46
def load_schema_file(path: str | Path) -> dict[str, Any]:
    """Load a JSON or YAML schema from a file, choosing by the `.json` extension.

    :param path: path to the schema file
    :type path: str | Path
    :return: the parsed schema document
    :rtype: dict[str, Any]
    """
    path = Path(path)
    loader = json.loads if path.name.endswith(".json") else yaml.safe_load
    return _require_mapping(loader(path.read_bytes()), f"from {path}")

coerce_extra_metadata_keywords 

coerce_extra_metadata_keywords(value: object) -> object

Normalize supported metadata keyword collections into a frozenset.

Parameters:

Name Type Description Default
value object

candidate keyword collection or raw input value

 required

Returns:

Type Description
object

normalized frozenset when value is a set, list, or tuple, else value unchanged
Source code in src/cdm_data_loaders/converters/core/io.py 
49
50
51
52
53
54
55
56
57
def coerce_extra_metadata_keywords(value: object) -> object:
    """Normalize supported metadata keyword collections into a frozenset.

    :param value: candidate keyword collection or raw input value
    :type value: object
    :return: normalized ``frozenset`` when ``value`` is a ``set``, ``list``, or ``tuple``, else ``value`` unchanged
    :rtype: object
    """
    return frozenset(value) if isinstance(value, (set, list, tuple)) else value