File: element.py

package info (click to toggle)
python-xsdata 24.1-2
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 2,936 kB
  • sloc: python: 29,257; xml: 404; makefile: 27; sh: 6
file content (460 lines) | stat: -rw-r--r-- 15,079 bytes parent folder | download
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
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
from typing import Any, Dict, List, Optional, Set, Type

from xsdata.exceptions import ParserError
from xsdata.formats.converter import converter
from xsdata.formats.dataclass.context import XmlContext
from xsdata.formats.dataclass.models.elements import XmlMeta, XmlVar
from xsdata.formats.dataclass.parsers import nodes
from xsdata.formats.dataclass.parsers.config import ParserConfig
from xsdata.formats.dataclass.parsers.mixins import XmlNode
from xsdata.formats.dataclass.parsers.utils import ParserUtils, PendingCollection
from xsdata.logger import logger
from xsdata.models.enums import DataType, Namespace
from xsdata.utils.namespaces import target_uri


class ElementNode(XmlNode):
    """
    XmlNode for complex elements and dataclasses.

    :param meta: Model xml metadata
    :param attrs: Key-value attribute mapping
    :param ns_map: Namespace prefix-URI map
    :param config: Parser configuration
    :param context: Model context provider
    :param position: The node position of objects cache
    :param mixed: The node supports mixed content
    :param derived_factory: Derived element factory
    :param xsi_type: The xml type substitution
    :param xsi_nil: The xml type substitution
    """

    __slots__ = (
        "meta",
        "attrs",
        "ns_map",
        "config",
        "context",
        "position",
        "mixed",
        "derived_factory",
        "xsi_type",
        "xsi_nil",
        "assigned",
        "tail_processed",
    )

    def __init__(
        self,
        meta: XmlMeta,
        attrs: Dict,
        ns_map: Dict,
        config: ParserConfig,
        context: XmlContext,
        position: int,
        mixed: bool = False,
        derived_factory: Optional[Type] = None,
        xsi_type: Optional[str] = None,
        xsi_nil: Optional[bool] = None,
    ):
        self.meta = meta
        self.attrs = attrs
        self.ns_map = ns_map
        self.config = config
        self.context = context
        self.position = position
        self.mixed = mixed
        self.derived_factory = derived_factory
        self.xsi_type = xsi_type
        self.xsi_nil = xsi_nil
        self.assigned: Set[int] = set()
        self.tail_processed: bool = False

    def bind(
        self, qname: str, text: Optional[str], tail: Optional[str], objects: List
    ) -> bool:
        obj: Any = None
        if not self.xsi_nil or self.meta.nillable:
            params: Dict = {}
            self.bind_attrs(params)
            self.bind_content(params, text, tail, objects)
            obj = self.config.class_factory(self.meta.clazz, params)

        if self.derived_factory:
            obj = self.derived_factory(qname=qname, value=obj, type=self.xsi_type)

        objects.append((qname, obj))

        if self.mixed and not self.tail_processed:
            tail = ParserUtils.normalize_content(tail)
            if tail:
                objects.append((None, tail))

        return True

    def bind_content(
        self, params: Dict, text: Optional[str], tail: Optional[str], objects: List[Any]
    ):
        wild_var = self.meta.find_any_wildcard()
        if wild_var and wild_var.mixed:
            self.bind_mixed_objects(params, wild_var, objects)
            bind_text = False
        else:
            self.bind_objects(params, objects)
            bind_text = self.bind_text(params, text)

        if not bind_text and wild_var:
            self.bind_wild_text(params, wild_var, text, tail)
            self.tail_processed = True

        for key in params:
            if isinstance(params[key], PendingCollection):
                params[key] = params[key].evaluate()

    def bind_attrs(self, params: Dict):
        """Parse the given element's attributes and any text content and return
        a dictionary of field names and values based on the given class
        metadata."""

        if not self.attrs:
            return

        for qname, value in self.attrs.items():
            var = self.meta.find_attribute(qname)
            if var and var.name not in params:
                self.bind_attr(params, var, value)
            else:
                var = self.meta.find_any_attributes(qname)
                if var:
                    self.bind_any_attr(params, var, qname, value)
                else:
                    if (
                        self.config.fail_on_unknown_attributes
                        and target_uri(qname) != Namespace.XSI.uri
                    ):
                        raise ParserError(
                            f"Unknown attribute {self.meta.qname}:{qname}"
                        )

    def bind_attr(self, params: Dict, var: XmlVar, value: Any):
        if var.init:
            params[var.name] = ParserUtils.parse_value(
                value=value,
                types=var.types,
                default=var.default,
                ns_map=self.ns_map,
                tokens_factory=var.tokens_factory,
                format=var.format,
            )

    def bind_any_attr(self, params: Dict, var: XmlVar, qname: str, value: Any):
        if var.name not in params:
            params[var.name] = {}

        params[var.name][qname] = ParserUtils.parse_any_attribute(value, self.ns_map)

    def bind_objects(self, params: Dict, objects: List):
        """Return a dictionary of qualified object names and their values for
        the given queue item."""

        position = self.position
        for qname, value in objects[position:]:
            if not self.bind_object(params, qname, value):
                logger.warning("Unassigned parsed object %s", qname)

        del objects[position:]

    def bind_object(self, params: Dict, qname: str, value: Any) -> bool:
        for var in self.meta.find_children(qname):
            if var.is_wildcard:
                return self.bind_wild_var(params, var, qname, value)

            if self.bind_var(params, var, value):
                return True

        return False

    @classmethod
    def bind_var(cls, params: Dict, var: XmlVar, value: Any) -> bool:
        """
        Add the given value to the params dictionary with the var name as key.

        Wrap the value to a list if var is a list. If the var name
        already exists it means we have a name conflict and the parser
        needs to lookup for any available wildcard fields.

        :return: Whether the binding process was successful or not.
        """
        if var.init:
            if var.list_element:
                items = params.get(var.name)
                if items is None:
                    params[var.name] = PendingCollection([value], var.factory)
                else:
                    items.append(value)
            elif var.name not in params:
                params[var.name] = value
            else:
                return False

        return True

    def bind_wild_var(self, params: Dict, var: XmlVar, qname: str, value: Any) -> bool:
        """
        Add the given value to the params dictionary with the wildcard var name
        as key.

        If the key is already present wrap the previous value into a
        generic AnyElement instance. If the previous value is already a
        generic instance add the current value as a child object.
        """
        value = self.prepare_generic_value(qname, value, var)

        if var.list_element:
            items = params.get(var.name)
            if items is None:
                params[var.name] = PendingCollection([value], var.factory)
            else:
                items.append(value)
        elif var.name in params:
            previous = params[var.name]
            factory = self.context.class_type.any_element

            if not isinstance(previous, factory) or previous.qname:
                params[var.name] = factory(children=[previous])

            params[var.name].children.append(value)
        else:
            params[var.name] = value

        return True

    def bind_mixed_objects(self, params: Dict, var: XmlVar, objects: List):
        """Return a dictionary of qualified object names and their values for
        the given mixed content xml var."""

        pos = self.position
        params[var.name] = [
            self.prepare_generic_value(qname, value, var)
            for qname, value in objects[pos:]
        ]
        del objects[pos:]

    def prepare_generic_value(
        self, qname: Optional[str], value: Any, var: XmlVar
    ) -> Any:
        """Prepare parsed value before binding to a wildcard field."""

        if qname and not self.context.class_type.is_model(value):
            any_factory = self.context.class_type.any_element
            value = any_factory(qname=qname, text=converter.serialize(value))

        return value

    def bind_text(self, params: Dict, text: Optional[str]) -> bool:
        """
        Add the given element's text content if any to the params dictionary
        with the text var name as key.

        Return if any data was bound.
        """
        var = self.meta.text

        if not var or (text is None and not self.xsi_nil):
            return False

        if var.init:
            if self.xsi_nil and not text:
                params[var.name] = None
            else:
                params[var.name] = ParserUtils.parse_value(
                    value=text,
                    types=var.types,
                    default=var.default,
                    ns_map=self.ns_map,
                    tokens_factory=var.tokens_factory,
                    format=var.format,
                )
        return True

    def bind_wild_text(
        self, params: Dict, var: XmlVar, txt: Optional[str], tail: Optional[str]
    ) -> bool:
        """
        Extract the text and tail content and bind it accordingly in the params
        dictionary. Return if any data was bound.

        - var is a list prepend the text and append the tail.
        - var is present in the params assign the text and tail to the generic object.
        - Otherwise bind the given element to a new generic object.
        """

        txt = ParserUtils.normalize_content(txt)
        tail = ParserUtils.normalize_content(tail)
        if txt is None and tail is None:
            return False

        if var.list_element:
            items = params.get(var.name)
            if items is None:
                params[var.name] = items = PendingCollection(None, var.factory)

            items.insert(0, txt)
            if tail:
                items.append(tail)

        else:
            previous = params.get(var.name, None)
            factory = self.context.class_type.any_element
            generic = factory(
                text=txt,
                tail=tail,
                attributes=ParserUtils.parse_any_attributes(self.attrs, self.ns_map),
            )
            if previous:
                generic.children.append(previous)

            params[var.name] = generic

        return True

    def child(self, qname: str, attrs: Dict, ns_map: Dict, position: int) -> XmlNode:
        for var in self.meta.find_children(qname):
            unique = 0 if not var.is_element or var.list_element else var.index
            if not unique or unique not in self.assigned:
                node = self.build_node(qname, var, attrs, ns_map, position)

                if node:
                    if unique:
                        self.assigned.add(unique)

                    return node

        if self.config.fail_on_unknown_properties:
            raise ParserError(f"Unknown property {self.meta.qname}:{qname}")

        return nodes.SkipNode()

    def build_node(
        self, qname: str, var: XmlVar, attrs: Dict, ns_map: Dict, position: int
    ) -> Optional[XmlNode]:
        if var.is_clazz_union:
            return nodes.UnionNode(
                var=var,
                attrs=attrs,
                ns_map=ns_map,
                config=self.config,
                context=self.context,
                position=position,
            )

        xsi_type = ParserUtils.xsi_type(attrs, ns_map)
        xsi_nil = ParserUtils.xsi_nil(attrs)
        derived_factory = self.context.class_type.derived_element

        if var.clazz:
            return self.build_element_node(
                var.clazz,
                var.derived,
                var.nillable,
                attrs,
                ns_map,
                position,
                derived_factory,
                xsi_type,
                xsi_nil,
            )

        if not var.any_type and not var.is_wildcard:
            return nodes.PrimitiveNode(
                var, ns_map, self.meta.mixed_content, derived_factory
            )

        datatype = DataType.from_qname(xsi_type) if xsi_type else None
        derived = var.derived or var.is_wildcard
        if datatype:
            return nodes.StandardNode(
                datatype, ns_map, var.nillable, derived_factory if derived else None
            )

        node = None
        clazz = None
        if xsi_type:
            clazz = self.context.find_type(xsi_type)

        if clazz:
            node = self.build_element_node(
                clazz,
                derived,
                var.nillable,
                attrs,
                ns_map,
                position,
                derived_factory,
                xsi_type,
                xsi_nil,
            )

        if node:
            return node

        if var.process_contents != "skip":
            clazz = self.context.find_type(qname)

        if clazz:
            node = self.build_element_node(
                clazz,
                False,
                var.nillable,
                attrs,
                ns_map,
                position,
                None,
                xsi_type,
                xsi_nil,
            )

        if node:
            return node

        return nodes.WildcardNode(
            var=var,
            attrs=attrs,
            ns_map=ns_map,
            position=position,
            factory=self.context.class_type.any_element,
        )

    def build_element_node(
        self,
        clazz: Type,
        derived: bool,
        nillable: bool,
        attrs: Dict,
        ns_map: Dict,
        position: int,
        derived_factory: Type,
        xsi_type: Optional[str] = None,
        xsi_nil: Optional[bool] = None,
    ) -> Optional[XmlNode]:
        meta = self.context.fetch(clazz, self.meta.namespace, xsi_type)
        nillable = nillable or meta.nillable

        if not meta or (xsi_nil is not None and nillable != xsi_nil):
            return None

        if xsi_type and not derived and not issubclass(meta.clazz, clazz):
            derived = True

        return ElementNode(
            meta=meta,
            config=self.config,
            attrs=attrs,
            ns_map=ns_map,
            context=self.context,
            position=position,
            derived_factory=derived_factory if derived else None,
            xsi_type=xsi_type,
            xsi_nil=xsi_nil,
            mixed=self.meta.mixed_content,
        )