"""
Provide the primitives which the modules reading and writing XML share.
Reading means walking a stream of ``(event, element)`` tuples: checking that
an element resides in the expected namespace, consuming the end element which
corresponds to a start element and taking the text out of an element. Writing
means framing an element. Each of these is defined here exactly once, so that
neither :py:mod:`aas_core3_2.xmlization` nor the XML-RPC subset
over which a JSON-able value is de/serialized carries a copy of its own.
The namespace is a constant of this module, :py:attr:`NAMESPACE`, and no
argument of its primitives. This module is generated for a single meta-model,
and a meta-model prescribes exactly one namespace, so there is nothing for
a caller to choose.
"""
# This code has been automatically generated by aas-core-codegen.
# Do NOT edit or append.
import json
import re
import sys
from typing import (
Callable,
Iterator,
List,
Mapping,
Optional,
Sequence,
TextIO,
Tuple,
Union,
)
if sys.version_info >= (3, 8):
from typing import Final, Literal, Protocol
else:
from typing_extensions import Final, Literal, Protocol
#: XML namespace in which all the elements of a document are expected to reside
NAMESPACE = "https://admin-shell.io/aas/3/2"
[docs]class Element(Protocol):
"""Behave like :py:meth:`xml.etree.ElementTree.Element`."""
@property
def attrib(self) -> Optional[Mapping[str, str]]:
"""Attributes of the element"""
raise NotImplementedError
@property
def text(self) -> Optional[str]:
"""Text content of the element"""
raise NotImplementedError
@property
def tail(self) -> Optional[str]:
"""Tail text of the element"""
raise NotImplementedError
@property
def tag(self) -> str:
"""Tag of the element; with a namespace provided as a ``{...}`` prefix"""
raise NotImplementedError
[docs] def clear(self) -> None:
"""Behave like :py:meth:`xml.etree.ElementTree.Element.clear`."""
raise NotImplementedError
[docs]class HasIterparse(Protocol):
"""Parse an XML document incrementally."""
# NOTE (mristin):
# ``self`` is not used in this context, but is necessary for Mypy,
# see: https://github.com/python/mypy/issues/5018 and
# https://github.com/python/mypy/commit/3efbc5c5e910296a60ed5b9e0e7eb11dd912c3ed#diff-e165eb7aed9dca0a5ebd93985c8cd263a6462d36ac185f9461348dc5a1396d76R9937
# NOTE (mristin):
# We require only the events which we actually use, and give them no default.
# Newer type stubs declare :py:func:`xml.etree.ElementTree.iterparse` as
# overloads over literal event names. A single overload must then match
# the protocol, which none would if the events were an optional sequence of
# arbitrary strings.
[docs] def iterparse(
self, source: TextIO, events: Sequence[Literal["start", "end"]]
) -> Iterator[Tuple[str, Element]]:
"""Behave like :py:func:`xml.etree.ElementTree.iterparse`."""
# NOTE (mristin):
# The SDK defines three of these path vocabularies: this one, the one in
# :py:mod:`aas_core3_2.jsonization`, and the one in
# :py:mod:`aas_core3_2.xmlcommon`.
# They look alike, and it is tempting to merge them, but they are not
# interchangeable.
#
# Each of them points into a different thing, and every segment carries
# a back-pointer to the container it stepped through, so the type of that
# back-pointer differs from one to the next:
#
# * This one points into the instances which you built, so a property
# segment holds a class of the meta-model.
# * The jsonization's points into the JSON-able structure being read or
# written, so a property segment holds the JSON-able mapping instead:
# while a document is being parsed, the instance which the property would
# belong to does not exist yet.
# * The xmlcommon's points into the XML document, where there are no
# properties at all, only elements, so it names an element instead.
#
# The three also render differently: a path into the instances is a Python
# access expression, a path into a JSON-able structure starts at the root of
# the document and carries no leading dot, and a path into an XML document
# is a relative XPath.
#
# Merging them would mean either dropping the back-pointers, which have been
# part of the public API of this SDK since before these modules were split
# apart, or defining a single path over the union of all the segment kinds --
# in which case every consumer would have to handle segments which can never
# occur in its world. Three small vocabularies which each say exactly what
# they can say cost less than one large one which lies about its range.
[docs]class ElementSegment:
"""Represent an element on a path to the erroneous value."""
#: Erroneous element
element: Final[Element]
[docs] def __init__(self, element: Element) -> None:
"""Initialize with the given values."""
self.element = element
[docs] def __str__(self) -> str:
"""
Render the segment as a tag without the namespace.
We deliberately omit the namespace in the tag names. If you want to actually
query with the resulting XPath, you have to insert the namespaces manually.
We did not know how to include the namespace in a meaningful way, as XPath
assumes namespace prefixes to be defined *outside* of the document. At least
the path thus rendered is informative, and you should be able to descend it
manually.
"""
_, has_namespace, tag_wo_ns = self.element.tag.rpartition("}")
if not has_namespace:
return self.element.tag
else:
return tag_wo_ns
[docs]class IndexSegment:
"""Represent an element in a sequence on a path to the erroneous value."""
#: Erroneous element
element: Final[Element]
#: Index of the element in the sequence, starting at zero
index: Final[int]
[docs] def __init__(self, element: Element, index: int) -> None:
"""Initialize with the given values."""
self.element = element
self.index = index
[docs] def __str__(self) -> str:
"""
Render the segment as an element wildcard with the index.
XPath counts the elements starting at one, so we add one to :py:attr:`index`.
"""
return f"*[{self.index + 1}]"
[docs]class KeySegment:
"""
Represent a member of an open JSON-able object on a path to
the erroneous element.
Unlike an :py:class:`ElementSegment`, which names an element prescribed by
the meta-model, a key is known only at run time, and can be any string
at all.
"""
#: Element of the ``<member>`` which carries the key
element: Final[Element]
#: Key of the member
key: Final[str]
[docs] def __init__(self, element: Element, key: str) -> None:
"""Initialize with the given values."""
self.element = element
self.key = key
[docs] def __str__(self) -> str:
"""
Render the segment as a predicate on the ``<name>`` child element.
A JSON-able object is written as an XML-RPC ``<struct>``, which carries
the key of a member in a ``<name>`` child element instead of in
an attribute, so the XPath has to match on that child element.
"""
escaped = (
self.key.replace("&", "&")
.replace("/", "/")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
.replace("'", "'")
)
return f'member[name="{escaped}"]'
Segment = Union[ElementSegment, IndexSegment, KeySegment]
[docs]class Path:
"""Represent the relative path to the erroneous element."""
[docs] def __init__(self) -> None:
"""Initialize as an empty path."""
self._segments: List[Segment] = []
@property
def segments(self) -> Sequence[Segment]:
"""Get the segments of the path."""
return self._segments
def _prepend(self, segment: Segment) -> None:
"""Insert the :paramref:`segment` in front of other segments."""
self._segments.insert(0, segment)
[docs] def __str__(self) -> str:
"""Render the path as a relative XPath.
We omit the leading ``/`` so that you can easily prefix it as you need.
"""
return "/".join(str(segment) for segment in self._segments)
[docs]class DeserializationException(Exception):
"""Signal that the XML de-serialization could not be performed."""
#: Human-readable explanation of the exception's cause
cause: Final[str]
#: Relative path to the erroneous value
path: Final[Path]
[docs] def __init__(self, cause: str) -> None:
"""Initialize with the given :paramref:`cause` and an empty path."""
self.cause = cause
self.path = Path()
[docs]class SerializationException(Exception):
"""Signal that the XML serialization could not be performed."""
#: Human-readable explanation of the exception's cause
cause: Final[str]
[docs] def __init__(self, cause: str) -> None:
"""Initialize with the given :paramref:`cause` and an empty path."""
self.cause = cause
self._segments: List[str] = []
@property
def path(self) -> str:
"""
Render the path to the erroneous value as a Python access expression.
The path points into the instance which you handed over for
the serialization, and *not* into an XML document -- at the point of
the failure, there is no document yet. For example, ``.submodels[0].id``
tells you that the serialization broke on ``that.submodels[0].id``.
Mind that the elements which the XML representation adds on top of
the instance contribute no segment, as they correspond to no attribute
access. This concerns the element enclosing the instance itself, and
the element which designates the model type of the value of a property.
"""
return "".join(self._segments)
def _prepend_property(self, name: str) -> None:
"""Insert the access to the property :paramref:`name` before the path."""
self._segments.insert(0, f".{name}")
def _prepend_index(self, index: int) -> None:
"""Insert the access to the item at :paramref:`index` before the path."""
self._segments.insert(0, f"[{index}]")
def _prepend_key(self, key: str) -> None:
"""
Insert the access to the member :paramref:`key` before the path.
Unlike a property of one of our classes, a member of an open JSON-able
object is known only at run time and can be any string at all, so it is
always rendered as a subscript.
"""
self._segments.insert(0, f"[{json.dumps(key, ensure_ascii=False)}]")
[docs] def __str__(self) -> str:
if len(self._segments) == 0:
return self.cause
return f"{self.path}: {self.cause}"
# region De-serialization
[docs]def parse_element_tag(element: Element) -> str:
"""
Extract the tag name without the namespace prefix from :paramref:`element`.
:param element: whose tag without namespace we want to extract
:return: tag name without the namespace prefix
:raise: :py:class:`DeserializationException` if unexpected :paramref:`element`
"""
# NOTE (mristin):
# :py:mod:`xml.etree.ElementTree` spells the tag of a namespaced element
# as ``{namespace}tag``. We match that prefix in place, with
# :py:meth:`str.startswith` and an offset, instead of composing
# the ``{...}`` form: this runs for every element of every document, and
# composing it would allocate a string on each call.
if (
element.tag.startswith("{")
and element.tag.startswith(NAMESPACE, 1)
and element.tag.startswith("}", len(NAMESPACE) + 1)
):
return element.tag[len(NAMESPACE) + 2 :]
got_namespace, separator, tag_wo_ns = element.tag.rpartition("}")
if separator:
if got_namespace.startswith("{"):
got_namespace = got_namespace[1:]
raise DeserializationException(
f"Expected the element in the namespace {NAMESPACE!r}, "
f"but got the element {tag_wo_ns!r} in "
f"the namespace {got_namespace!r}"
)
raise DeserializationException(
f"Expected the element in the namespace {NAMESPACE!r}, "
f"but got the element {tag_wo_ns!r} without the namespace prefix"
)
[docs]def parse_unqualified_element_tag(element: Element) -> str:
"""
Extract the tag name of :paramref:`element`, which is expected to reside in
no namespace at all.
:param element: whose tag we want to extract
:return: tag name
:raise: :py:class:`DeserializationException` if unexpected :paramref:`element`
"""
if not element.tag.startswith("{"):
return element.tag
got_namespace, _, tag_wo_ns = element.tag[1:].partition("}")
raise DeserializationException(
f"Expected the element in no namespace, "
f"but got the element {tag_wo_ns!r} in "
f"the namespace {got_namespace!r}"
)
[docs]def raise_if_non_whitespace_text(element: Element, expected_what: str) -> None:
"""
Check that :paramref:`element` encloses no text of its own.
An element which holds other elements may still be spelled over several
lines, so the whitespace between its children is text as far as
:py:mod:`xml.etree.ElementTree` is concerned. Anything else is not.
:param element: to be verified
:param expected_what: what the element was expected to enclose
:raise:
:py:class:`.DeserializationException` if there is text;
conforming to the convention about handling error paths,
the exception path is left empty.
"""
if element.text is not None and len(element.text.strip()) != 0:
raise DeserializationException(
f"Expected only {expected_what} and whitespace text, "
f"but got text: {element.text!r}"
)
[docs]def read_next_start_element(
iterator: Iterator[Tuple[str, Element]], expected_what: str
) -> Element:
"""
Read the next start element from :paramref:`iterator`.
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:param expected_what: what we expected to read, for the error messages
:raise:
:py:class:`.DeserializationException` if the input ended, or if what came
next was no start element; conforming to the convention about handling
error paths, the exception path is left empty.
:return: the start element
"""
next_event_element = next(iterator, None)
if next_event_element is None:
raise DeserializationException(
f"Expected the start element for {expected_what}, "
f"but got the end-of-input"
)
next_event, next_element = next_event_element
if next_event != "start":
raise DeserializationException(
f"Expected the start element for {expected_what}, "
f"but got event {next_event!r} and element {next_element.tag!r}"
)
return next_element
[docs]def raise_if_has_tail_or_attrib(element: Element) -> None:
"""
Check that :paramref:`element` has no trailing text and no attributes.
:param element: to be verified
:raise:
:py:class:`.DeserializationException` if trailing text or attributes;
conforming to the convention about handling error paths,
the exception path is left empty.
"""
if element.tail is not None and len(element.tail.strip()) != 0:
raise DeserializationException(
f"Expected no trailing text, but got: {element.tail!r}"
)
if element.attrib is not None and len(element.attrib) > 0:
raise DeserializationException(
f"Expected no attributes, but got: {element.attrib}"
)
[docs]def read_end_element(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> Element:
"""
Read the end element corresponding to the start :paramref:`element`
from :paramref:`iterator`.
:param element: corresponding start element
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
"""
next_event_element = next(iterator, None)
if next_event_element is None:
raise DeserializationException(
f"Expected the end element for {element.tag}, " f"but got the end-of-input"
)
next_event, next_element = next_event_element
if next_event != "end" or next_element.tag != element.tag:
raise DeserializationException(
f"Expected the end element for {element.tag!r}, "
f"but got the event {next_event!r} and element {next_element.tag!r}"
)
raise_if_has_tail_or_attrib(next_element)
return next_element
[docs]def read_text_from_element(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> str:
"""
Extract the text from the :paramref:`element`, and read
the end element from :paramref:`iterator`.
The :paramref:`element` is expected to contain text. Otherwise,
it is considered as unexpected input.
:param element: start element enclosing the text
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
"""
raise_if_has_tail_or_attrib(element)
text = element.text
end_element = read_end_element(
element,
iterator,
)
if text is None:
if end_element.text is None:
raise DeserializationException(
"Expected an element with text, but got an element with no text."
)
text = end_element.text
return text
[docs]def read_str_from_element_text(
element: Element, iterator: Iterator[Tuple[str, Element]]
) -> str:
"""
Parse the text of :paramref:`element` as a string, and
read the corresponding end element from :paramref:`iterator`.
If there is no text, empty string is returned.
:param element: start element
:param iterator:
Input stream of ``(event, element)`` coming from
:py:func:`xml.etree.ElementTree.iterparse` with the argument
``events=["start", "end"]``
:raise: :py:class:`DeserializationException` if unexpected input
:return: parsed value
"""
# NOTE (mristin):
# We do not use ``read_text_from_element`` as that function expects
# the ``element`` to contain *some* text. In contrast, this function
# can also deal with empty text, in which case it returns an empty string.
text = element.text
end_element = read_end_element(element, iterator)
if text is None:
text = end_element.text
raise_if_has_tail_or_attrib(element)
return text if text is not None else ""
#: Match a run of the whitespace characters which XSD knows
XS_WHITESPACE_RE = re.compile(r"[ \t\n\r]+")
[docs]def collapse_whitespace(text: str) -> str:
"""
Normalize :paramref:`text` the way ``whiteSpace="collapse"`` prescribes.
Every atomic XSD type except a string, and every type derived from one
by restriction, fixes ``whiteSpace`` to ``collapse``, and a schema author
can not change it. A tab, a line feed and a carriage return each become
a space, a run of spaces becomes one space, and the leading and trailing
spaces go. Only then is the result a lexical representation to be matched.
Mind that this strips only the whitespace *around* the value: a space
within it survives as a single space, so ``2 3`` becomes ``2 3``, which
is still no number.
See: https://www.w3.org/TR/xmlschema-2/#rf-whiteSpace
:param text: to be normalized
:return: normalized text
"""
return XS_WHITESPACE_RE.sub(" ", text).strip(" ")
# endregion
# region Serialization
[docs]class Writer:
"""
Write the XML elements of a single document to :py:attr:`~stream`.
The namespace is specified on the very first element alone, and on none of
the elements which follow it, so one writer serves exactly one document.
"""
#: Stream to be written to
stream: Final[TextIO]
#: Number of the currently open elements which reside in no namespace
_open_in_no_namespace: int
#: Method pointer to be invoked for writing the start element with or without
#: specifying a namespace (depending on the state of the writer)
write_start_element: Callable[[str], None]
#: Method pointer to be invoked for writing an empty element with or without
#: specifying a namespace (depending on the state of the writer)
write_empty_element: Callable[[str], None]
# NOTE (mristin):
# The serialization procedure is quite rigid. We leverage the specifics of
# the serialization procedure to optimize the code a bit.
#
# Namely, we model the writing of the XML elements as a state machine.
# The namespace is only specified for the very first element. All the subsequent
# elements will *not* have the namespace specified. We implement that behavior by
# using pointers to methods, as Python treats the methods as first-class citizens.
#
# The ``write_start_element`` will point to
# ``_write_first_start_element_with_namespace`` on the *first* invocation.
# Afterwards, it will be redirected to ``_write_start_element_without_namespace``.
#
# Analogously for ``write_empty_element``.
#
# Please see the implementation for the details, but this should give you at least
# a rough overview.
def _write_first_start_element_with_namespace(self, name: str) -> None:
"""
Write the start element with the tag name :paramref:`name` and specify
its namespace.
The :py:attr:`~write_start_element` is set to
:py:meth:`~_write_start_element_without_namespace` after the first invocation
of this method.
:param name: of the element tag. Expected to contain no XML special characters.
"""
self.stream.write(f'<{name} xmlns="{NAMESPACE}">')
# NOTE (mristin):
# Any subsequence call to `write_start_element` or `write_empty_element`
# should not specify the namespace of the element as we specified now already
# specified it.
self.write_start_element = self._write_start_element_without_namespace
self.write_empty_element = self._write_empty_element_without_namespace
def _write_start_element_without_namespace(self, name: str) -> None:
"""
Write the start element with the tag name :paramref:`name`.
The first element, written *before* this one, is expected to have been
already written with the namespace specified.
:param name: of the element tag. Expected to contain no XML special characters.
"""
self.stream.write(f"<{name}>")
[docs] def write_end_element(self, name: str) -> None:
"""
Write the end element with the tag name :paramref:`name`.
:param name: of the element tag. Expected to contain no XML special characters.
"""
self.stream.write(f"</{name}>")
[docs] def write_start_element_in_no_namespace(self, name: str) -> None:
"""
Write the start element with the tag name :paramref:`name`, which resides
in no namespace at all.
The outermost such element undeclares the default namespace, so that
the elements nested in it reside in no namespace as well and need no
declaration of their own.
:param name: of the element tag. Expected to contain no XML special characters.
"""
if self._open_in_no_namespace == 0:
self.stream.write(f'<{name} xmlns="">')
else:
self.stream.write(f"<{name}>")
self._open_in_no_namespace += 1
[docs] def write_end_element_in_no_namespace(self, name: str) -> None:
"""
Write the end element with the tag name :paramref:`name`, which resides in
no namespace at all.
:param name: of the element tag. Expected to contain no XML special characters.
"""
self._open_in_no_namespace -= 1
self.stream.write(f"</{name}>")
[docs] def write_empty_element_in_no_namespace(self, name: str) -> None:
"""
Write the empty element with the tag name :paramref:`name`, which resides
in no namespace at all.
If it is the outermost such element, it undeclares the default namespace.
:param name: of the element tag. Expected to contain no XML special characters.
"""
if self._open_in_no_namespace == 0:
self.stream.write(f'<{name} xmlns=""/>')
else:
self.stream.write(f"<{name}/>")
def _write_first_empty_element_with_namespace(self, name: str) -> None:
"""
Write the first (and only) empty element with the tag name :paramref:`name`.
No elements are expected to be written to the stream afterwards. The element
includes the namespace specification.
:param name: of the element tag. Expected to contain no XML special characters.
"""
self.stream.write(f'<{name} xmlns="{NAMESPACE}"/>')
self.write_empty_element = self._raise_if_write_element_called_again
self.write_start_element = self._raise_if_write_element_called_again
def _raise_if_write_element_called_again(self, name: str) -> None:
raise AssertionError(
f"We expected to call ``_write_first_empty_element_with_namespace`` "
f"only once. This is an unexpected second call for writing "
f"an (empty or non-empty) element with the tag name: {name!r}"
)
def _write_empty_element_without_namespace(self, name: str) -> None:
"""
Write the empty element with the tag name :paramref:`name`.
The call to this method is expected to occur *after* the enclosing element with
a specified namespace has been written.
:param name: of the element tag. Expected to contain no XML special characters.
"""
self.stream.write(f"<{name}/>")
[docs] def __init__(self, stream: TextIO) -> None:
"""
Initialize the writer to write to :paramref:`stream`.
The first element will include the :py:attr:`NAMESPACE`. Every other
element will not have the namespace specified.
:param stream: where to write to
"""
self.stream = stream
self._open_in_no_namespace = 0
self.write_start_element = self._write_first_start_element_with_namespace
self.write_empty_element = self._write_first_empty_element_with_namespace
# endregion