cdm_data_loaders.converters.readers.xsd ¶

@dataclass(frozen=True, slots=True)
class XsdReader:
    """Build typed converter IR documents from XML Schema Definition (XSD) files.

    :param xml_tag: Default target XML element tag name when not supplied to :meth:`read`.
    :type xml_tag: str | None
    :param max_recursion_depth: Maximum times a recursive ``XsdComplexType`` may be expanded along an ancestor path.
    :type max_recursion_depth: int
    :param error_type: Base exception class raised when a construct cannot be converted.
    :type error_type: type[ConversionError]
    :param error_by_code: Mapping from specification error code (``E1``-``E9``) to exception class.
    :type error_by_code: Mapping[str, type[ConversionError]]
    """

    xml_tag: str | None = None
    max_recursion_depth: int = DEFAULT_MAX_RECURSION_DEPTH
    error_type: type[ConversionError] = ConversionError
    error_by_code: Mapping[str, type[ConversionError]] = field(default_factory=frozendict)

    def __post_init__(self) -> None:
        """Validate reader configuration."""
        validate_max_recursion_depth(self.max_recursion_depth)

    def _raise_err(self, code: str, msg: str) -> NoReturn:
        """Raise the configured exception subclass for ``code`` or fall back to ``error_type``.

        :param code: Specification error code (``E1``-``E9``).
        :type code: str
        :param msg: Diagnostic error message.
        :type msg: str
        :raises ConversionError: Always raised with the resolved error class.
        """
        _raise_code(code, msg, error_type=self.error_type, error_by_code=self.error_by_code)

    def read(
        self,
        source: xmlschema.XMLSchema | str | Path,
        xml_tag: str | None = None,
        *,
        struct_cache: dict[tuple[int, bool], Any] | None = None,
    ) -> SchemaDocument:
        """Read an ``xmlschema.XMLSchema`` element into a ``SchemaDocument``.

        :param source: Loaded ``XMLSchema`` instance or filesystem path to an XSD file.
        :type source: xmlschema.XMLSchema | str | Path
        :param xml_tag: Target XML element name (local name or Clark notation). Defaults to ``self.xml_tag`` or the sole root element.
        :type xml_tag: str | None
        :param struct_cache: Optional cache sharing IR struct ``TypedNode``s across calls.
        :type struct_cache: dict[tuple[int, bool], Any] | None
        :return: Structural IR document with ``source="xsd"`` provenance.
        :rtype: SchemaDocument
        :raises ConversionError: If the schema or ``xml_tag`` violates ``E1``-``E9``.
        """
        schema = load_schema(source) if isinstance(source, str | Path) else source
        tag = xml_tag if xml_tag is not None else self.xml_tag
        if tag is None:
            roots = resolve_root_elements(schema, error_type=self.error_type, error_by_code=self.error_by_code)
            if len(roots) != 1 or roots[0].local_name is None:
                msg = "xml_tag must be specified when the XSD schema does not have a single root element"
                self._raise_err("E1", msg)
            tag = roots[0].local_name

        target_elem, _, ctx = _prepare_target_traversal(
            schema,
            tag,
            max_recursion_depth=self.max_recursion_depth,
            struct_cache=struct_cache,
            error_type=self.error_type,
            error_by_code=self.error_by_code,
        )
        root_struct = _build_element_struct_node(target_elem, tag, ctx)
        return _finish_element_doc(schema, target_elem, tag, root_struct)

    def read_at_level(
        self,
        source: xmlschema.XMLSchema | str | Path,
        level: int,
        *,
        root_tag: str | None = None,
    ) -> dict[str, SchemaDocument]:
        """Read all XML element types at depth ``level`` in an XSD schema into ``SchemaDocument``s.

        :param source: Loaded ``XMLSchema`` instance or filesystem path to an XSD file.
        :type source: xmlschema.XMLSchema | str | Path
        :param level: Non-negative depth relative to the root element(s).
        :type level: int
        :param root_tag: Optional explicit root element name to anchor the traversal.
        :type root_tag: str | None
        :return: Mapping of element local tag name to its ``SchemaDocument``.
        :rtype: dict[str, SchemaDocument]
        """
        schema = load_schema(source) if isinstance(source, str | Path) else source
        elements_at_level = resolve_elements_at_level(
            schema,
            level,
            root_tag=root_tag,
            error_type=self.error_type,
            error_by_code=self.error_by_code,
        )
        struct_cache: dict[tuple[int, bool], Any] = {}
        return {
            tag_name: _build_element_table_doc(
                schema,
                target_elem,
                tag_name,
                max_recursion_depth=self.max_recursion_depth,
                struct_cache=struct_cache,
                error_type=self.error_type,
                error_by_code=self.error_by_code,
            )
            for tag_name, target_elem in elements_at_level.items()
        }