cdm_data_loaders.coercion.xml ¶

def make_xml_element_converter(
    schema: xmlschema.XMLSchema,
    xml_tag: str,
    *,
    max_recursion_depth: int = DEFAULT_MAX_RECURSION_DEPTH,
) -> Callable[[Element], dict[str, Any]]:
    """Build a reusable element converter for ``xml_tag`` that resolves the XSD target once.

    :param schema: Loaded XMLSchema instance.
    :type schema: xmlschema.XMLSchema
    :param xml_tag: Target XML element name (local name or Clark-notation ``{ns}local``).
    :type xml_tag: str
    :param max_recursion_depth: Maximum structural recursion depth before JSON serialization.
    :type max_recursion_depth: int
    :return: Function converting an lxml Element into a sparse dictionary.
    :rtype: Callable[[Element], dict[str, Any]]
    """
    _, elem_type, ctx = _prepare_target_traversal(
        schema,
        xml_tag,
        max_recursion_depth=max_recursion_depth,
        error_type=XsdToPyArrowError,
        error_by_code=ERROR_BY_CODE,
    )
    return functools.partial(_convert_struct_element_to_dict, ctx=ctx, elem_type=elem_type)

def make_xml_level_converters(
    schema: xmlschema.XMLSchema,
    level: int,
    *,
    root_tag: str | None = None,
    max_recursion_depth: int = DEFAULT_MAX_RECURSION_DEPTH,
) -> dict[str, Callable[[Element], dict[str, Any]]]:
    """Build reusable element converters for all XML element types at depth ``level`` in an XSD schema.

    :param schema: Loaded XMLSchema instance.
    :type schema: xmlschema.XMLSchema
    :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 max_recursion_depth: Maximum structural recursion depth before JSON serialization.
    :type max_recursion_depth: int
    :return: Mapping of element local tag name to its element-to-dict converter function.
    :rtype: dict[str, Callable[[Element], dict[str, Any]]]
    """
    elements_at_level = resolve_elements_at_level(
        schema,
        level,
        root_tag=root_tag,
        error_type=XsdToPyArrowError,
        error_by_code=ERROR_BY_CODE,
    )
    return {
        tag_name: _make_resolved_element_converter(
            schema,
            target_elem,
            tag_name,
            max_recursion_depth=max_recursion_depth,
        )
        for tag_name, target_elem in elements_at_level.items()
    }

def convert_xml_element(
    element: Element,
    schema: xmlschema.XMLSchema,
    xml_tag: str,
    *,
    max_recursion_depth: int = DEFAULT_MAX_RECURSION_DEPTH,
) -> dict[str, Any]:
    """Convert an lxml Element matching ``xml_tag`` into a typed dictionary according to the XSD.

    :param element: XML element to convert.
    :type element: Element
    :param schema: Loaded XMLSchema instance.
    :type schema: xmlschema.XMLSchema
    :param xml_tag: Target XML element name (local name or Clark-notation ``{ns}local``).
    :type xml_tag: str
    :param max_recursion_depth: Maximum structural recursion depth before JSON serialization.
    :type max_recursion_depth: int
    :return: Sparse dictionary of coerced field values matching the generated PyArrow schema.
    :rtype: dict[str, Any]
    """
    return make_xml_element_converter(
        schema,
        xml_tag,
        max_recursion_depth=max_recursion_depth,
    )(element)