Skip to content

cdm_data_loaders.converters.facades.dlt_to_jsonschema

Convert dlt schemas through the typed reader and JSON Schema emitter.

DltToJSONSchemaError

Bases: ConversionError

Raised when a dlt schema construct cannot be converted.

Source code in src/cdm_data_loaders/converters/facades/dlt_to_jsonschema.py 
11
12
class DltToJSONSchemaError(ConversionError):
    """Raised when a dlt schema construct cannot be converted."""

DltToJSONSchema pydantic-model

Bases: JsonSchemaConverterBase[dict[str, dict[str, Any]]]

Read root table documents and emit their JSON Schema equivalents.

Fields:

  • extension_registry (InstanceOf[ExtensionRegistry])
  • child_table_mode (Literal['object', 'array'])
  • include_dlt_columns (bool)
  • include_variant_columns (bool)
  • preserve_unknown_hints (bool)
Source code in src/cdm_data_loaders/converters/facades/dlt_to_jsonschema.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class DltToJSONSchema(JsonSchemaConverterBase[dict[str, dict[str, Any]]]):
    """Read root table documents and emit their JSON Schema equivalents."""

    _allow_yaml_in_string: ClassVar[bool] = True

    child_table_mode: Literal["object", "array"] = "object"
    include_dlt_columns: bool = False
    include_variant_columns: bool = False
    preserve_unknown_hints: bool = True

    def convert(self, schema: dict[str, Any]) -> dict[str, dict[str, Any]]:
        """Convert a stored schema or table mapping into root JSON documents."""
        reader = DltReader(
            include_dlt_columns=self.include_dlt_columns,
            include_variant_columns=self.include_variant_columns,
            child_table_mode=self.child_table_mode,
            extension_registry=self.extension_registry,
        )
        emitter = JsonSchemaEmitter(preserve_unknown_hints=self.preserve_unknown_hints)
        try:
            return {name: emitter.emit(document) for name, document in reader.read(schema).items()}
        except ConversionError as error:
            raise DltToJSONSchemaError(str(error)) from error

convert 

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

Convert a stored schema or table mapping into root JSON documents.

Source code in src/cdm_data_loaders/converters/facades/dlt_to_jsonschema.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
def convert(self, schema: dict[str, Any]) -> dict[str, dict[str, Any]]:
    """Convert a stored schema or table mapping into root JSON documents."""
    reader = DltReader(
        include_dlt_columns=self.include_dlt_columns,
        include_variant_columns=self.include_variant_columns,
        child_table_mode=self.child_table_mode,
        extension_registry=self.extension_registry,
    )
    emitter = JsonSchemaEmitter(preserve_unknown_hints=self.preserve_unknown_hints)
    try:
        return {name: emitter.emit(document) for name, document in reader.read(schema).items()}
    except ConversionError as error:
        raise DltToJSONSchemaError(str(error)) from error