1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
|
import copy
import warnings
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple, Type, cast
from xsdata.exceptions import ConverterWarning, ParserError
from xsdata.formats.bindings import T
from xsdata.formats.dataclass.context import XmlContext
from xsdata.formats.dataclass.parsers.config import ParserConfig
from xsdata.formats.dataclass.parsers.mixins import (
EventsHandler,
PushParser,
XmlHandler,
XmlNode,
)
from xsdata.formats.dataclass.parsers.utils import ParserUtils
from xsdata.models.enums import EventType
Parsed = Tuple[Optional[str], Any]
@dataclass
class NodeParser(PushParser):
"""
Bind xml nodes to dataclasses.
:param config: Parser configuration
:param context: Model context provider
:param handler: Override default XmlHandler
:ivar ms_map: Namespace registry of parsed prefix-URI mappings
"""
config: ParserConfig = field(default_factory=ParserConfig)
context: XmlContext = field(default_factory=XmlContext)
handler: Type[XmlHandler] = field(default=EventsHandler)
ns_map: Dict = field(init=False, default_factory=dict)
def parse(self, source: Any, clazz: Optional[Type[T]] = None) -> T:
"""Parse the input stream or filename and return the resulting object
tree."""
handler = self.handler(clazz=clazz, parser=self)
with warnings.catch_warnings():
if self.config.fail_on_converter_warnings:
warnings.filterwarnings("error", category=ConverterWarning)
try:
result = handler.parse(source)
except (ConverterWarning, SyntaxError) as e:
raise ParserError(e)
if result is not None:
return result
target_class = clazz.__name__ if clazz else ""
raise ParserError(f"Failed to create target class `{target_class}`")
def start(
self,
clazz: Optional[Type],
queue: List[XmlNode],
objects: List[Parsed],
qname: str,
attrs: Dict,
ns_map: Dict,
):
"""
Start element notification receiver.
Build and queue the XmlNode for the starting element.
:param clazz: Root class type, if it's missing look for any
suitable models from the current context.
:param queue: The active XmlNode queue
:param objects: The list of all intermediate parsed objects
:param qname: Qualified name
:param attrs: Attribute key-value map
:param ns_map: Namespace prefix-URI map
"""
from xsdata.formats.dataclass.parsers.nodes import ElementNode, WrapperNode
try:
item = queue[-1]
if isinstance(item, ElementNode) and qname in item.meta.wrappers:
child = cast(XmlNode, WrapperNode(parent=item))
else:
child = item.child(qname, attrs, ns_map, len(objects))
except IndexError:
xsi_type = ParserUtils.xsi_type(attrs, ns_map)
# Match element qname directly
if clazz is None:
clazz = self.context.find_type(qname)
# Root is xs:anyType try xsi:type
if clazz is None and xsi_type:
clazz = self.context.find_type(xsi_type)
# Exit if we still have no binding model
if clazz is None:
raise ParserError(f"No class found matching root: {qname}")
meta = self.context.fetch(clazz, xsi_type=xsi_type)
if xsi_type is None or meta.qname == qname:
derived_factory = None
else:
derived_factory = self.context.class_type.derived_element
xsi_nil = ParserUtils.xsi_nil(attrs)
child = ElementNode(
position=0,
meta=meta,
config=self.config,
attrs=attrs,
ns_map=ns_map,
context=self.context,
derived_factory=derived_factory,
xsi_type=xsi_type if derived_factory else None,
xsi_nil=xsi_nil,
)
queue.append(child)
def end(
self,
queue: List[XmlNode],
objects: List[Parsed],
qname: str,
text: Optional[str],
tail: Optional[str],
) -> bool:
"""
End element notification receiver.
Pop the last XmlNode from the queue and use it to build and
return the resulting object tree with its text and tail content.
:param queue: Xml nodes queue
:param objects: List of parsed objects
:param qname: Qualified name
:param text: Text content
:param tail: Tail content
"""
item = queue.pop()
return item.bind(qname, text, tail, objects)
@dataclass
class RecordParser(NodeParser):
"""
Bind xml nodes to dataclasses and store the intermediate events.
:ivar events: List of pushed events
"""
events: List = field(init=False, default_factory=list)
def start(
self,
clazz: Optional[Type],
queue: List[XmlNode],
objects: List[Parsed],
qname: str,
attrs: Dict,
ns_map: Dict,
):
"""
Start element notification receiver.
Build and queue the XmlNode for the starting element, append the
event with the attributes and ns map to the events list.
:param clazz: Root class type, if it's missing look for any
suitable models from the current context.
:param queue: The active XmlNode queue
:param objects: The list of all intermediate parsed objects
:param qname: Qualified name
:param attrs: Attributes key-value map
:param ns_map: Namespace prefix-URI map
"""
self.events.append((EventType.START, qname, copy.deepcopy(attrs), ns_map))
super().start(clazz, queue, objects, qname, attrs, ns_map)
def end(
self,
queue: List[XmlNode],
objects: List[Parsed],
qname: str,
text: Optional[str],
tail: Optional[str],
) -> Any:
"""
End element notification receiver.
Pop the last XmlNode from the queue and use it to build and
return the resulting object tree with its text and tail content.
Append the end event with the text,tail content to the events
list.
:param queue: Xml nodes queue
:param objects: List of parsed objects
:param qname: Qualified name
:param text: Text content
:param tail: Tail content
"""
self.events.append((EventType.END, qname, text, tail))
return super().end(queue, objects, qname, text, tail)
def register_namespace(self, prefix: Optional[str], uri: str):
self.events.append((EventType.START_NS, prefix, uri))
super().register_namespace(prefix, uri)
|