cdm_data_loaders.converters.facades.xsd_to_pyarrow ¶

class XsdToPyArrowError(ConversionError):
    """Base class for XSD constructs that cannot be converted to a PyArrow schema.

    Each subclass carries a stable specification ``code`` and a short ``title``; its docstring states the reason
    the construct is rejected. The converter specification document is generated from these attributes.
    """

    code: ClassVar[str] = ""
    title: ClassVar[str] = ""

class XmlTagNotFoundError(XsdToPyArrowError):
    """The target element is not declared anywhere in the XSD."""

    code = "E1"
    title = "`xml_tag` not found"

class AmbiguousXmlTagError(XsdToPyArrowError):
    """Declarations of `xml_tag` at the shallowest matching depth have different types, so no single `pa.Schema` can be chosen."""

    code = "E2"
    title = "Ambiguous `xml_tag` type"

class NonStructXmlTagError(XsdToPyArrowError):
    """A `pa.Table` schema needs top-level columns: the type of `xml_tag` must yield at least one attribute, `#text`, or child element field."""

    code = "E3"
    title = "Non-struct top-level `xml_tag`"

class WildcardElementError(XsdToPyArrowError):
    """Arbitrary undeclared child elements cannot be represented as fixed `pa.struct` fields."""

    code = "E4"
    title = "Element wildcard (`xs:any`)"

class WildcardAttributeError(XsdToPyArrowError):
    """Arbitrary undeclared attributes cannot be represented as fixed `pa.struct` fields."""

    code = "E5"
    title = "Attribute wildcard (`xs:anyAttribute`)"

class AbstractComponentError(XsdToPyArrowError):
    """Concrete instances of an abstract element or type require `xsi:type` or substitution polymorphism, which has no single schema."""

    code = "E6"
    title = 'Abstract element or type (`abstract="true"`)'

class SubstitutionGroupError(XsdToPyArrowError):
    """Substituted elements introduce element names and types that are not declared in the parent content model."""

    code = "E7"
    title = "Substitution group"

class FieldNameCollisionError(XsdToPyArrowError):
    """An attribute (prefixed `_`) and a child element map to the same field name in one `pa.struct`."""

    code = "E8"
    title = "Attribute / child field name collision"

class RecursiveNonJsonTypeError(XsdToPyArrowError):
    """Subtrees truncated beyond `max_recursion_depth` are serialized with `json.dumps`, which cannot encode `datetime.date` or `bytes` without lossy conversion."""

    code = "E9"
    title = "Non-JSON-serializable scalar under recursion"

class XsdToPyArrow(BaseModel):
    """Facade converting XML Schema Definition (XSD) documents into PyArrow schemas and empty tables."""

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

    xml_tag: str | None = None
    max_recursion_depth: PositiveInt = Field(default=DEFAULT_MAX_RECURSION_DEPTH)

    def _read(self, schema: xmlschema.XMLSchema | str | Path, xml_tag: str | None = None) -> SchemaDocument:
        """Read an XSD schema element into a converter IR ``SchemaDocument``.

        :param schema: Loaded ``XMLSchema`` instance or path to an XSD file.
        :type schema: xmlschema.XMLSchema | str | Path
        :param xml_tag: Target XML element name.
        :type xml_tag: str | None
        :return: Parsed ``SchemaDocument``.
        :rtype: SchemaDocument
        """
        return XsdReader(
            xml_tag=self.xml_tag,
            max_recursion_depth=self.max_recursion_depth,
            error_type=XsdToPyArrowError,
            error_by_code=ERROR_BY_CODE,
        ).read(schema, xml_tag)

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

        :param schema: Loaded ``XMLSchema`` instance or path to an XSD file.
        :type schema: 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 tag name to ``SchemaDocument``.
        :rtype: dict[str, SchemaDocument]
        """
        return XsdReader(
            max_recursion_depth=self.max_recursion_depth,
            error_type=XsdToPyArrowError,
            error_by_code=ERROR_BY_CODE,
        ).read_at_level(schema, level, root_tag=root_tag)

    def convert(self, schema: xmlschema.XMLSchema | str | Path, xml_tag: str | None = None) -> pa.Schema:
        """Convert an XSD schema element into a ``pa.Schema``.

        :param schema: loaded ``XMLSchema`` instance or path to an XSD file
        :type schema: xmlschema.XMLSchema | str | Path
        :param xml_tag: target XML element name (defaults to ``self.xml_tag`` or sole root element)
        :type xml_tag: str | None
        :return: converted PyArrow schema
        :rtype: pa.Schema
        """
        return DEFAULT_EMITTER.emit(self._read(schema, xml_tag))

    def convert_to_table(self, schema: xmlschema.XMLSchema | str | Path, xml_tag: str | None = None) -> pa.Table:
        """Convert an XSD schema element into a zero-row ``pa.Table``.

        :param schema: loaded ``XMLSchema`` instance or path to an XSD file
        :type schema: xmlschema.XMLSchema | str | Path
        :param xml_tag: target XML element name (defaults to ``self.xml_tag`` or sole root element)
        :type xml_tag: str | None
        :return: zero-row PyArrow table with the converted schema
        :rtype: pa.Table
        """
        return DEFAULT_EMITTER.emit_table(self._read(schema, xml_tag))

    def convert_to_python(
        self,
        schema: xmlschema.XMLSchema | str | Path,
        xml_tag: str | None = None,
        *,
        docstring: str | None = None,
    ) -> str:
        """Convert an XSD schema element into a static Python schema module.

        :param schema: loaded ``XMLSchema`` instance or path to an XSD file
        :type schema: xmlschema.XMLSchema | str | Path
        :param xml_tag: target XML element name (defaults to ``self.xml_tag`` or sole root element)
        :type xml_tag: str | None
        :param docstring: optional module docstring header
        :type docstring: str | None
        :return: formatted Python source code defining ``*_STRUCT``, ``*_SCHEMA``, ``SCHEMA``, and ``SCHEMAS``
        :rtype: str
        """
        return DEFAULT_EMITTER.emit_python(self._read(schema, xml_tag), docstring=docstring)

    def convert_to_python_file(
        self,
        schema: xmlschema.XMLSchema | str | Path,
        output: str | Path,
        xml_tag: str | None = None,
        *,
        docstring: str | None = None,
    ) -> Path:
        """Convert an XSD schema element and write a static Python schema module to ``output``.

        :param schema: loaded ``XMLSchema`` instance or path to an XSD file
        :type schema: xmlschema.XMLSchema | str | Path
        :param output: destination file path for the generated Python module
        :type output: str | Path
        :param xml_tag: target XML element name (defaults to ``self.xml_tag`` or sole root element)
        :type xml_tag: str | None
        :param docstring: optional module docstring header
        :type docstring: str | None
        :return: resolved output ``Path``
        :rtype: Path
        """
        return DEFAULT_EMITTER.emit_python_file(self._read(schema, xml_tag), output, docstring=docstring)

    def convert_level_to_python(
        self,
        schema: xmlschema.XMLSchema | str | Path,
        level: int = 1,
        *,
        root_tag: str | None = None,
        docstring: str | None = None,
    ) -> str:
        """Convert all XML element types at depth ``level`` in an XSD schema into a static Python schema module.

        :param schema: loaded ``XMLSchema`` instance or path to an XSD file
        :type schema: 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
        :param docstring: optional module docstring header
        :type docstring: str | None
        :return: formatted Python source code defining ``*_STRUCT``, ``*_SCHEMA``, ``SCHEMA``, and ``SCHEMAS``
        :rtype: str
        """
        return DEFAULT_EMITTER.emit_python(self._read_level(schema, level, root_tag=root_tag), docstring=docstring)

    def convert_level_to_python_file(
        self,
        schema: xmlschema.XMLSchema | str | Path,
        output: str | Path,
        level: int = 1,
        *,
        root_tag: str | None = None,
        docstring: str | None = None,
    ) -> Path:
        """Convert all XML element types at depth ``level`` in an XSD schema and write a Python module to ``output``.

        :param schema: loaded ``XMLSchema`` instance or path to an XSD file
        :type schema: xmlschema.XMLSchema | str | Path
        :param output: destination file path for the generated Python module
        :type output: 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
        :param docstring: optional module docstring header
        :type docstring: str | None
        :return: resolved output ``Path``
        :rtype: Path
        """
        return DEFAULT_EMITTER.emit_python_file(
            self._read_level(schema, level, root_tag=root_tag),
            output,
            docstring=docstring,
        )