Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# pylint: disable=line-too-long,useless-suppression,too-many-lines
2# coding=utf-8
3# --------------------------------------------------------------------------
4# Copyright (c) Microsoft Corporation. All rights reserved.
5# Licensed under the MIT License. See License.txt in the project root for license information.
6# Code generated by Microsoft (R) Python Code Generator.
7# Changes may cause incorrect behavior and will be lost if the code is regenerated.
8# --------------------------------------------------------------------------
9# pylint: disable=protected-access, broad-except
11import copy
12import calendar
13import decimal
14import functools
15import sys
16import logging
17import base64
18import re
19import typing
20import enum
21import email.utils
22from datetime import datetime, date, time, timedelta, timezone
23from json import JSONEncoder
24import xml.etree.ElementTree as ET
25from collections.abc import MutableMapping
26import isodate
27from azure.core.exceptions import DeserializationError
28from azure.core import CaseInsensitiveEnumMeta
29from azure.core.pipeline import PipelineResponse
30from azure.core.serialization import _Null
32from azure.core.rest import HttpResponse
34if sys.version_info >= (3, 11):
35 from typing import Self
36else:
37 from typing_extensions import Self
39_LOGGER = logging.getLogger(__name__)
41__all__ = ["SdkJSONEncoder", "Model", "rest_field", "rest_discriminator"]
43TZ_UTC = timezone.utc
44_T = typing.TypeVar("_T")
45_NONE_TYPE = type(None)
48def _timedelta_as_isostr(td: timedelta) -> str:
49 """Converts a datetime.timedelta object into an ISO 8601 formatted string, e.g. 'P4DT12H30M05S'
51 Function adapted from the Tin Can Python project: https://github.com/RusticiSoftware/TinCanPython
53 :param timedelta td: The timedelta to convert
54 :rtype: str
55 :return: ISO8601 version of this timedelta
56 """
58 # Split seconds to larger units
59 seconds = td.total_seconds()
60 minutes, seconds = divmod(seconds, 60)
61 hours, minutes = divmod(minutes, 60)
62 days, hours = divmod(hours, 24)
64 days, hours, minutes = list(map(int, (days, hours, minutes)))
65 seconds = round(seconds, 6)
67 # Build date
68 date_str = ""
69 if days:
70 date_str = "%sD" % days
72 if hours or minutes or seconds:
73 # Build time
74 time_str = "T"
76 # Hours
77 bigger_exists = date_str or hours
78 if bigger_exists:
79 time_str += "{:02}H".format(hours)
81 # Minutes
82 bigger_exists = bigger_exists or minutes
83 if bigger_exists:
84 time_str += "{:02}M".format(minutes)
86 # Seconds
87 try:
88 if seconds.is_integer():
89 seconds_string = "{:02}".format(int(seconds))
90 else:
91 # 9 chars long w/ leading 0, 6 digits after decimal
92 seconds_string = "%09.6f" % seconds
93 # Remove trailing zeros
94 seconds_string = seconds_string.rstrip("0")
95 except AttributeError: # int.is_integer() raises
96 seconds_string = "{:02}".format(seconds)
98 time_str += "{}S".format(seconds_string)
99 else:
100 time_str = ""
102 return "P" + date_str + time_str
105def _serialize_bytes(o, format: typing.Optional[str] = None) -> str:
106 encoded = base64.b64encode(o).decode()
107 if format == "base64url":
108 return encoded.strip("=").replace("+", "-").replace("/", "_")
109 return encoded
112def _serialize_duration(td: timedelta, format: typing.Optional[str] = None):
113 """Serialize a timedelta to its wire representation.
115 For the ``seconds``/``milliseconds`` encodings the value is converted to a
116 numeric value, otherwise it falls back to an ISO 8601 duration string.
118 :param timedelta td: The timedelta to serialize.
119 :param str format: The duration encoding format.
120 :rtype: int or float or str
121 :return: serialized duration
122 """
123 seconds = td.total_seconds()
124 if format == "duration-seconds-int":
125 return int(seconds)
126 if format == "duration-seconds-float":
127 return seconds
128 if format == "duration-milliseconds-int":
129 return int(seconds * 1000)
130 if format == "duration-milliseconds-float":
131 return seconds * 1000
132 return _timedelta_as_isostr(td)
135def _serialize_datetime(o, format: typing.Optional[str] = None):
136 if hasattr(o, "year") and hasattr(o, "hour"):
137 if format == "rfc7231":
138 return email.utils.format_datetime(o, usegmt=True)
139 if format == "unix-timestamp":
140 return int(calendar.timegm(o.utctimetuple()))
142 # astimezone() fails for naive times in Python 2.7, so make make sure o is aware (tzinfo is set)
143 if not o.tzinfo:
144 iso_formatted = o.replace(tzinfo=TZ_UTC).isoformat()
145 else:
146 iso_formatted = o.astimezone(TZ_UTC).isoformat()
147 # Replace the trailing "+00:00" UTC offset with "Z" (RFC 3339: https://www.ietf.org/rfc/rfc3339.txt)
148 return iso_formatted.replace("+00:00", "Z")
149 # Next try datetime.date or datetime.time
150 return o.isoformat()
153def _is_readonly(p):
154 try:
155 return p._visibility == ["read"]
156 except AttributeError:
157 return False
160class SdkJSONEncoder(JSONEncoder):
161 """A JSON encoder that's capable of serializing datetime objects and bytes."""
163 def __init__(self, *args, exclude_readonly: bool = False, format: typing.Optional[str] = None, **kwargs):
164 super().__init__(*args, **kwargs)
165 self.exclude_readonly = exclude_readonly
166 self.format = format
168 def default(self, o): # pylint: disable=too-many-return-statements
169 if _is_model(o):
170 if self.exclude_readonly:
171 readonly_props = [p._rest_name for p in o._attr_to_rest_field.values() if _is_readonly(p)]
172 return {k: v for k, v in o.items() if k not in readonly_props}
173 return dict(o.items())
174 try:
175 return super(SdkJSONEncoder, self).default(o)
176 except TypeError:
177 if isinstance(o, _Null):
178 return None
179 if isinstance(o, decimal.Decimal):
180 return float(o)
181 if isinstance(o, (bytes, bytearray)):
182 return _serialize_bytes(o, self.format)
183 try:
184 # First try datetime.datetime
185 return _serialize_datetime(o, self.format)
186 except AttributeError:
187 pass
188 # Last, try datetime.timedelta
189 try:
190 return _timedelta_as_isostr(o)
191 except AttributeError:
192 # This will be raised when it hits value.total_seconds in the method above
193 pass
194 return super(SdkJSONEncoder, self).default(o)
197_VALID_DATE = re.compile(r"\d{4}[-]\d{2}[-]\d{2}T\d{2}:\d{2}:\d{2}" + r"\.?\d*Z?[-+]?[\d{2}]?:?[\d{2}]?")
198_VALID_RFC7231 = re.compile(
199 r"(Mon|Tue|Wed|Thu|Fri|Sat|Sun),\s\d{2}\s"
200 r"(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s\d{4}\s\d{2}:\d{2}:\d{2}\sGMT"
201)
203_ARRAY_ENCODE_MAPPING = {
204 "pipeDelimited": "|",
205 "spaceDelimited": " ",
206 "commaDelimited": ",",
207 "newlineDelimited": "\n",
208}
211def _deserialize_array_encoded(delimit: str, attr):
212 if isinstance(attr, str):
213 if attr == "":
214 return []
215 return attr.split(delimit)
216 return attr
219def _deserialize_datetime(attr: typing.Union[str, datetime]) -> datetime:
220 """Deserialize ISO-8601 formatted string into Datetime object.
222 :param str attr: response string to be deserialized.
223 :rtype: ~datetime.datetime
224 :returns: The datetime object from that input
225 """
226 if isinstance(attr, datetime):
227 # i'm already deserialized
228 return attr
229 attr = attr.upper()
230 match = _VALID_DATE.match(attr)
231 if not match:
232 raise ValueError("Invalid datetime string: " + attr)
234 check_decimal = attr.split(".")
235 if len(check_decimal) > 1:
236 decimal_str = ""
237 for digit in check_decimal[1]:
238 if digit.isdigit():
239 decimal_str += digit
240 else:
241 break
242 if len(decimal_str) > 6:
243 attr = attr.replace(decimal_str, decimal_str[0:6])
245 date_obj = isodate.parse_datetime(attr)
246 test_utc = date_obj.utctimetuple()
247 if test_utc.tm_year > 9999 or test_utc.tm_year < 1:
248 raise OverflowError("Hit max or min date")
249 return date_obj # type: ignore[no-any-return]
252def _deserialize_datetime_rfc7231(attr: typing.Union[str, datetime]) -> datetime:
253 """Deserialize RFC7231 formatted string into Datetime object.
255 :param str attr: response string to be deserialized.
256 :rtype: ~datetime.datetime
257 :returns: The datetime object from that input
258 """
259 if isinstance(attr, datetime):
260 # i'm already deserialized
261 return attr
262 match = _VALID_RFC7231.match(attr)
263 if not match:
264 raise ValueError("Invalid datetime string: " + attr)
266 return email.utils.parsedate_to_datetime(attr)
269def _deserialize_datetime_unix_timestamp(attr: typing.Union[float, datetime]) -> datetime:
270 """Deserialize unix timestamp into Datetime object.
272 :param str attr: response string to be deserialized.
273 :rtype: ~datetime.datetime
274 :returns: The datetime object from that input
275 """
276 if isinstance(attr, datetime):
277 # i'm already deserialized
278 return attr
279 return datetime.fromtimestamp(attr, TZ_UTC)
282def _deserialize_date(attr: typing.Union[str, date]) -> date:
283 """Deserialize ISO-8601 formatted string into Date object.
284 :param str attr: response string to be deserialized.
285 :rtype: date
286 :returns: The date object from that input
287 """
288 # This must NOT use defaultmonth/defaultday. Using None ensure this raises an exception.
289 if isinstance(attr, date):
290 return attr
291 return isodate.parse_date(attr, defaultmonth=None, defaultday=None) # type: ignore
294def _deserialize_time(attr: typing.Union[str, time]) -> time:
295 """Deserialize ISO-8601 formatted string into time object.
297 :param str attr: response string to be deserialized.
298 :rtype: datetime.time
299 :returns: The time object from that input
300 """
301 if isinstance(attr, time):
302 return attr
303 return isodate.parse_time(attr) # type: ignore[no-any-return]
306def _deserialize_bytes(attr):
307 if isinstance(attr, (bytes, bytearray)):
308 return attr
309 return bytes(base64.b64decode(attr))
312def _deserialize_bytes_base64(attr):
313 if isinstance(attr, (bytes, bytearray)):
314 return attr
315 padding = "=" * (3 - (len(attr) + 3) % 4) # type: ignore
316 attr = attr + padding # type: ignore
317 encoded = attr.replace("-", "+").replace("_", "/")
318 return bytes(base64.b64decode(encoded))
321def _deserialize_duration(attr):
322 if isinstance(attr, timedelta):
323 return attr
324 return isodate.parse_duration(attr)
327def _deserialize_duration_numeric(attr, unit):
328 if isinstance(attr, timedelta):
329 return attr
330 return timedelta(**{unit: float(attr)})
333def _deserialize_decimal(attr):
334 if isinstance(attr, decimal.Decimal):
335 return attr
336 return decimal.Decimal(str(attr))
339def _deserialize_int_as_str(attr):
340 if isinstance(attr, int):
341 return attr
342 return int(attr)
345_DESERIALIZE_MAPPING = {
346 datetime: _deserialize_datetime,
347 date: _deserialize_date,
348 time: _deserialize_time,
349 bytes: _deserialize_bytes,
350 bytearray: _deserialize_bytes,
351 timedelta: _deserialize_duration,
352 typing.Any: lambda x: x,
353 decimal.Decimal: _deserialize_decimal,
354}
356_DESERIALIZE_MAPPING_WITHFORMAT = {
357 "rfc3339": _deserialize_datetime,
358 "rfc7231": _deserialize_datetime_rfc7231,
359 "unix-timestamp": _deserialize_datetime_unix_timestamp,
360 "base64": _deserialize_bytes,
361 "base64url": _deserialize_bytes_base64,
362 "duration-seconds-int": functools.partial(_deserialize_duration_numeric, unit="seconds"),
363 "duration-seconds-float": functools.partial(_deserialize_duration_numeric, unit="seconds"),
364 "duration-milliseconds-int": functools.partial(_deserialize_duration_numeric, unit="milliseconds"),
365 "duration-milliseconds-float": functools.partial(_deserialize_duration_numeric, unit="milliseconds"),
366}
369def get_deserializer(annotation: typing.Any, rf: typing.Optional["_RestField"] = None):
370 if annotation is int and rf and rf._format == "str":
371 return _deserialize_int_as_str
372 if annotation is str and rf and rf._format in _ARRAY_ENCODE_MAPPING:
373 return functools.partial(_deserialize_array_encoded, _ARRAY_ENCODE_MAPPING[rf._format])
374 if rf and rf._format:
375 return _DESERIALIZE_MAPPING_WITHFORMAT.get(rf._format)
376 return _DESERIALIZE_MAPPING.get(annotation) # pyright: ignore
379def _get_type_alias_type(module_name: str, alias_name: str):
380 types = {
381 k: v
382 for k, v in sys.modules[module_name].__dict__.items()
383 if isinstance(v, typing._GenericAlias) # type: ignore
384 }
385 if alias_name not in types:
386 return alias_name
387 return types[alias_name]
390def _get_model(module_name: str, model_name: str):
391 models = {k: v for k, v in sys.modules[module_name].__dict__.items() if isinstance(v, type)}
392 module_end = module_name.rsplit(".", 1)[0]
393 models.update({k: v for k, v in sys.modules[module_end].__dict__.items() if isinstance(v, type)})
394 if isinstance(model_name, str):
395 model_name = model_name.split(".")[-1]
396 if model_name not in models:
397 return model_name
398 return models[model_name]
401_UNSET = object()
404class _MyMutableMapping(MutableMapping[str, typing.Any]):
405 def __init__(self, data: dict[str, typing.Any]) -> None:
406 self._data = data
408 def __contains__(self, key: typing.Any) -> bool:
409 return key in self._data
411 def __getitem__(self, key: str) -> typing.Any:
412 # If this key has been deserialized (for mutable types), we need to handle serialization
413 if hasattr(self, "_attr_to_rest_field"):
414 cache_attr = f"_deserialized_{key}"
415 if hasattr(self, cache_attr):
416 rf = _get_rest_field(getattr(self, "_attr_to_rest_field"), key)
417 if rf:
418 value = self._data.get(key)
419 if isinstance(value, (dict, list, set)):
420 # For mutable types, serialize and return
421 # But also update _data with serialized form and clear flag
422 # so mutations via this returned value affect _data
423 serialized = _serialize(value, rf._format)
424 # If serialized form is same type (no transformation needed),
425 # return _data directly so mutations work
426 if isinstance(serialized, type(value)) and serialized == value:
427 return self._data.get(key)
428 # Otherwise return serialized copy and clear flag
429 try:
430 object.__delattr__(self, cache_attr)
431 except AttributeError:
432 pass
433 # Store serialized form back
434 self._data[key] = serialized
435 return serialized
436 return self._data.__getitem__(key)
438 def __setitem__(self, key: str, value: typing.Any) -> None:
439 # Clear any cached deserialized value when setting through dictionary access
440 cache_attr = f"_deserialized_{key}"
441 try:
442 object.__delattr__(self, cache_attr)
443 except AttributeError:
444 pass
445 self._data.__setitem__(key, value)
447 def __delitem__(self, key: str) -> None:
448 self._data.__delitem__(key)
450 def __iter__(self) -> typing.Iterator[typing.Any]:
451 return self._data.__iter__()
453 def __len__(self) -> int:
454 return self._data.__len__()
456 def __ne__(self, other: typing.Any) -> bool:
457 return not self.__eq__(other)
459 def keys(self) -> typing.KeysView[str]:
460 """
461 :returns: a set-like object providing a view on D's keys
462 :rtype: ~typing.KeysView
463 """
464 return self._data.keys()
466 def values(self) -> typing.ValuesView[typing.Any]:
467 """
468 :returns: an object providing a view on D's values
469 :rtype: ~typing.ValuesView
470 """
471 return self._data.values()
473 def items(self) -> typing.ItemsView[str, typing.Any]:
474 """
475 :returns: set-like object providing a view on D's items
476 :rtype: ~typing.ItemsView
477 """
478 return self._data.items()
480 def get(self, key: str, default: typing.Any = None) -> typing.Any:
481 """
482 Get the value for key if key is in the dictionary, else default.
483 :param str key: The key to look up.
484 :param any default: The value to return if key is not in the dictionary. Defaults to None
485 :returns: D[k] if k in D, else d.
486 :rtype: any
487 """
488 try:
489 return self[key]
490 except KeyError:
491 return default
493 @typing.overload
494 def pop(self, key: str) -> typing.Any: ... # pylint: disable=arguments-differ
496 @typing.overload
497 def pop(self, key: str, default: _T) -> _T: ... # pylint: disable=signature-differs
499 @typing.overload
500 def pop(self, key: str, default: typing.Any) -> typing.Any: ... # pylint: disable=signature-differs
502 def pop(self, key: str, default: typing.Any = _UNSET) -> typing.Any:
503 """
504 Removes specified key and return the corresponding value.
505 :param str key: The key to pop.
506 :param any default: The value to return if key is not in the dictionary
507 :returns: The value corresponding to the key.
508 :rtype: any
509 :raises KeyError: If key is not found and default is not given.
510 """
511 if default is _UNSET:
512 return self._data.pop(key)
513 return self._data.pop(key, default)
515 def popitem(self) -> tuple[str, typing.Any]:
516 """
517 Removes and returns some (key, value) pair
518 :returns: The (key, value) pair.
519 :rtype: tuple
520 :raises KeyError: if D is empty.
521 """
522 return self._data.popitem()
524 def clear(self) -> None:
525 """
526 Remove all items from D.
527 """
528 self._data.clear()
530 def update(self, *args: typing.Any, **kwargs: typing.Any) -> None: # pylint: disable=arguments-differ
531 """
532 Updates D from mapping/iterable E and F.
533 :param any args: Either a mapping object or an iterable of key-value pairs.
534 """
535 self._data.update(*args, **kwargs)
537 @typing.overload
538 def setdefault(self, key: str, default: None = None) -> None: ...
540 @typing.overload
541 def setdefault(self, key: str, default: typing.Any) -> typing.Any: ... # pylint: disable=signature-differs
543 def setdefault(self, key: str, default: typing.Any = _UNSET) -> typing.Any:
544 """
545 Same as calling D.get(k, d), and setting D[k]=d if k not found
546 :param str key: The key to look up.
547 :param any default: The value to set if key is not in the dictionary
548 :returns: D[k] if k in D, else d.
549 :rtype: any
550 """
551 if default is _UNSET:
552 return self._data.setdefault(key)
553 return self._data.setdefault(key, default)
555 def __eq__(self, other: typing.Any) -> bool:
556 if isinstance(other, _MyMutableMapping):
557 return self._data == other._data
558 try:
559 other_model = self.__class__(other)
560 except Exception:
561 return False
562 return self._data == other_model._data
564 def __repr__(self) -> str:
565 return str(self._data)
568def _is_model(obj: typing.Any) -> bool:
569 return getattr(obj, "_is_model", False)
572def _serialize(o, format: typing.Optional[str] = None): # pylint: disable=too-many-return-statements
573 if isinstance(o, list):
574 if format in _ARRAY_ENCODE_MAPPING and all(isinstance(x, str) for x in o):
575 return _ARRAY_ENCODE_MAPPING[format].join(o)
576 return [_serialize(x, format) for x in o]
577 if isinstance(o, dict):
578 return {k: _serialize(v, format) for k, v in o.items()}
579 if isinstance(o, set):
580 return {_serialize(x, format) for x in o}
581 if isinstance(o, tuple):
582 return tuple(_serialize(x, format) for x in o)
583 if isinstance(o, (bytes, bytearray)):
584 return _serialize_bytes(o, format)
585 if isinstance(o, decimal.Decimal):
586 return float(o)
587 if isinstance(o, enum.Enum):
588 return o.value
589 if isinstance(o, int):
590 if format == "str":
591 return str(o)
592 return o
593 try:
594 # First try datetime.datetime
595 return _serialize_datetime(o, format)
596 except AttributeError:
597 pass
598 # Last, try datetime.timedelta
599 try:
600 return _serialize_duration(o, format)
601 except AttributeError:
602 # This will be raised when it hits value.total_seconds in the method above
603 pass
604 return o
607def _get_rest_field(attr_to_rest_field: dict[str, "_RestField"], rest_name: str) -> typing.Optional["_RestField"]:
608 try:
609 return next(rf for rf in attr_to_rest_field.values() if rf._rest_name == rest_name)
610 except StopIteration:
611 return None
614def _create_value(rf: typing.Optional["_RestField"], value: typing.Any) -> typing.Any:
615 if not rf:
616 return _serialize(value, None)
617 if rf._is_multipart_file_input:
618 return value
619 if rf._is_model:
620 return _deserialize(rf._type, value)
621 if isinstance(value, ET.Element):
622 value = _deserialize(rf._type, value)
623 return _serialize(value, rf._format)
626# ============================================================================
627# Fast-path scalar deserializer functions for rest_field(deserializer=...)
628# These are referenced from rest_field declarations to bypass the generic
629# _deserialize -> _deserialize_with_callable chain.
630# Only simple/primitive types — no models or container types.
631# ============================================================================
634def _xml_deser_str(value):
635 if isinstance(value, ET.Element):
636 return value.text or ""
637 return str(value) if value is not None else None
640def _xml_deser_int(value):
641 if isinstance(value, ET.Element):
642 return int(value.text) if value.text else None
643 return int(value) if value is not None else None
646def _xml_deser_float(value):
647 if isinstance(value, ET.Element):
648 return float(value.text) if value.text else None
649 return float(value) if value is not None else None
652def _xml_deser_bool(value):
653 if isinstance(value, ET.Element):
654 text = value.text
655 else:
656 text = value
657 if text is None:
658 return None
659 if text in (True, False):
660 return text
661 return text.lower() == "true"
664# pylint: disable=docstring-missing-param
665def _xml_deser_bytes(value):
666 """Deserialize bytes from XML (base64)."""
667 if isinstance(value, ET.Element):
668 text = value.text
669 else:
670 text = value
671 if text is None:
672 return None
673 return _deserialize_bytes(text)
676def _xml_deser_bytes_base64url(value):
677 """Deserialize bytes from XML (base64url)."""
678 if isinstance(value, ET.Element):
679 text = value.text
680 else:
681 text = value
682 if text is None:
683 return None
684 return _deserialize_bytes_base64(text)
687def _xml_deser_datetime(value):
688 """Deserialize a datetime from XML (ISO 8601 / rfc3339)."""
689 if isinstance(value, ET.Element):
690 text = value.text
691 else:
692 text = value
693 if text is None:
694 return None
695 return _deserialize_datetime(text)
698def _xml_deser_datetime_rfc7231(value):
699 """Deserialize a datetime from XML (RFC7231 format)."""
700 if isinstance(value, ET.Element):
701 text = value.text
702 else:
703 text = value
704 if text is None:
705 return None
706 return _deserialize_datetime_rfc7231(text)
709def _xml_deser_datetime_unix_timestamp(value):
710 """Deserialize a datetime from XML (Unix timestamp)."""
711 if isinstance(value, ET.Element):
712 text = value.text
713 else:
714 text = value
715 if text is None:
716 return None
717 return _deserialize_datetime_unix_timestamp(float(text))
720def _xml_deser_date(value):
721 """Deserialize a date from XML (ISO 8601)."""
722 if isinstance(value, ET.Element):
723 text = value.text
724 else:
725 text = value
726 if text is None:
727 return None
728 return _deserialize_date(text)
731def _xml_deser_time(value):
732 """Deserialize a time from XML (ISO 8601)."""
733 if isinstance(value, ET.Element):
734 text = value.text
735 else:
736 text = value
737 if text is None:
738 return None
739 return _deserialize_time(text)
742def _xml_deser_duration(value):
743 """Deserialize a timedelta from XML (ISO 8601 duration)."""
744 if isinstance(value, ET.Element):
745 text = value.text
746 else:
747 text = value
748 if text is None:
749 return None
750 return _deserialize_duration(text)
753def _xml_deser_decimal(value):
754 """Deserialize a Decimal from XML."""
755 if isinstance(value, ET.Element):
756 text = value.text
757 else:
758 text = value
759 if text is None:
760 return None
761 return _deserialize_decimal(text)
764def _xml_deser_enum_or_str(enum_cls, value):
765 """Deserialize a Union[EnumType, str] from XML."""
766 text = value.text if isinstance(value, ET.Element) else value
767 if text is None:
768 return None
769 try:
770 return enum_cls(text)
771 except ValueError:
772 return text
775def _extract_xml_model_type(rf_type):
776 """Extract the concrete Model class from a resolved rf._type partial chain.
778 Unwraps ``Optional[Model]`` and ``_deserialize_model(Model, ...)``
779 wrappers. Only handles Model and Optional[Model] — other composite
780 types (List, Dict, Union, etc.) return None and fall through to the
781 generic ``_deserialize`` path at runtime.
782 """
783 if rf_type is None:
784 return None
785 if isinstance(rf_type, type) and _is_model(rf_type):
786 return rf_type
787 if not isinstance(rf_type, functools.partial):
788 return None
789 func = rf_type.func
790 args = rf_type.args
791 if func is _deserialize_with_optional and args:
792 return _extract_xml_model_type(args[0])
793 if func is _deserialize_model and args:
794 cls = args[0]
795 return cls if isinstance(cls, type) and _is_model(cls) else None
796 return None
799def _build_xml_field_plan( # pylint: disable=docstring-missing-return, docstring-missing-rtype, unused-variable
800 cls, attr_to_rest_field: dict
801) -> list:
802 """Build a precomputed XML field plan for fast _init_from_xml iteration.
804 Called once per model class in __new__. Returns a list of tuples:
805 (rest_name, xml_name, kind, deser, rf_type, is_optional, items_name)
807 kind: 0=wrapped, 1=attribute, 2=unwrapped, 3=text
809 For Model and Optional[Model] fields that lack a scalar
810 ``_deserializer``, this function precomputes the Model class as the
811 deserializer so ``_init_from_xml`` can call ``ModelClass(element)``
812 directly instead of going through the expensive
813 ``_get_deserialize_callable_from_annotation`` chain at runtime.
814 """
815 model_meta = getattr(cls, "_xml", {})
816 model_ns = model_meta.get("ns") or model_meta.get("namespace")
817 plan = []
819 for rf in attr_to_rest_field.values():
820 prop_meta = getattr(rf, "_xml", {})
821 deser = rf._deserializer
823 xml_name = prop_meta.get("name", rf._rest_name)
824 xml_ns = _resolve_xml_ns(prop_meta, model_meta)
825 if xml_ns:
826 xml_name = "{" + xml_ns + "}" + xml_name
828 is_optional = rf._is_optional
830 # For Model / Optional[Model] fields without a scalar deserializer,
831 # precompute the Model class as the deserializer.
832 if deser is None and rf._type is not None:
833 model_cls = _extract_xml_model_type(rf._type)
834 if model_cls is not None:
835 deser = model_cls
837 if prop_meta.get("attribute", False):
838 plan.append((rf._rest_name, xml_name, 1, deser, rf._type, is_optional, None))
839 elif prop_meta.get("unwrapped", False):
840 items_name = prop_meta.get("itemsName")
841 if items_name:
842 items_ns = prop_meta.get("itemsNs")
843 if items_ns is not None:
844 xml_ns = items_ns
845 if xml_ns:
846 items_name = "{" + xml_ns + "}" + items_name
847 else:
848 items_name = xml_name
849 plan.append((rf._rest_name, xml_name, 2, deser, rf._type, is_optional, items_name))
850 elif prop_meta.get("text", False):
851 plan.append((rf._rest_name, xml_name, 3, deser, rf._type, is_optional, None))
852 else:
853 plan.append((rf._rest_name, xml_name, 0, deser, rf._type, is_optional, None))
855 return plan
858# pylint: enable=docstring-missing-param
859class Model(_MyMutableMapping):
860 _is_model = True
861 # label whether current class's _attr_to_rest_field has been calculated
862 # could not see _attr_to_rest_field directly because subclass inherits it from parent class
863 _calculated: set[str] = set()
865 def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
866 class_name = self.__class__.__name__
867 if len(args) > 1:
868 raise TypeError(f"{class_name}.__init__() takes 2 positional arguments but {len(args) + 1} were given")
869 dict_to_pass: dict[str, typing.Any] = {}
870 if args:
871 if isinstance(args[0], ET.Element):
872 dict_to_pass.update(self._init_from_xml(args[0]))
873 else:
874 dict_to_pass.update(
875 {k: _create_value(_get_rest_field(self._attr_to_rest_field, k), v) for k, v in args[0].items()}
876 )
877 else:
878 non_attr_kwargs = [k for k in kwargs if k not in self._attr_to_rest_field]
879 if non_attr_kwargs:
880 # actual type errors only throw the first wrong keyword arg they see, so following that.
881 raise TypeError(f"{class_name}.__init__() got an unexpected keyword argument '{non_attr_kwargs[0]}'")
882 dict_to_pass.update(
883 {
884 self._attr_to_rest_field[k]._rest_name: _create_value(self._attr_to_rest_field[k], v)
885 for k, v in kwargs.items()
886 if v is not None
887 }
888 )
889 # Apply client default values for fields the caller didn't set so that
890 # defaults are part of `_data` and therefore included during serialization.
891 for rf in self._attr_to_rest_field.values():
892 if rf._default is _UNSET:
893 continue
894 if rf._rest_name in dict_to_pass:
895 continue
896 dict_to_pass[rf._rest_name] = _create_value(rf, rf._default)
897 super().__init__(dict_to_pass)
899 def _init_from_xml( # pylint: disable=too-many-branches, too-many-statements
900 self, element: ET.Element
901 ) -> dict[str, typing.Any]:
902 """Deserialize an XML element into a dict mapping rest field names to values.
904 :param ET.Element element: The XML element to deserialize from.
905 :returns: A dictionary of rest_name to deserialized value pairs.
906 :rtype: dict
907 """
908 result: dict[str, typing.Any] = {}
909 existed_attr_keys: list[str] = []
911 field_plan = getattr(self, "_xml_field_plan", None)
912 if field_plan:
913 for rest_name, xml_name, kind, deser, rf_type, is_optional, items_name in field_plan:
914 if kind == 0: # wrapped element (most common)
915 item = element.find(xml_name)
916 if item is not None:
917 existed_attr_keys.append(xml_name)
918 if deser:
919 result[rest_name] = deser(item)
920 else:
921 result[rest_name] = _deserialize(rf_type, item)
922 elif kind == 1: # attribute
923 attr_val = element.get(xml_name)
924 if attr_val is not None:
925 existed_attr_keys.append(xml_name)
926 if deser:
927 result[rest_name] = deser(attr_val)
928 else:
929 result[rest_name] = attr_val
930 elif kind == 2: # unwrapped array
931 items = element.findall(items_name) # pyright: ignore
932 if len(items) > 0:
933 existed_attr_keys.append(items_name)
934 if deser:
935 result[rest_name] = deser(items)
936 else:
937 result[rest_name] = _deserialize(rf_type, items)
938 elif not is_optional:
939 existed_attr_keys.append(items_name)
940 result[rest_name] = []
941 elif kind == 3: # text
942 if element.text is not None:
943 if deser:
944 result[rest_name] = deser(element.text)
945 else:
946 result[rest_name] = element.text
947 else:
948 model_meta = getattr(self, "_xml", {})
949 for rf in self._attr_to_rest_field.values():
950 prop_meta = getattr(rf, "_xml", {})
951 xml_name = prop_meta.get("name", rf._rest_name)
952 xml_ns = _resolve_xml_ns(prop_meta, model_meta)
953 if xml_ns:
954 xml_name = "{" + xml_ns + "}" + xml_name
956 # attribute
957 if prop_meta.get("attribute", False) and element.get(xml_name) is not None:
958 existed_attr_keys.append(xml_name)
959 result[rf._rest_name] = _deserialize(rf._type, element.get(xml_name))
960 continue
962 # unwrapped element is array
963 if prop_meta.get("unwrapped", False):
964 _items_name = prop_meta.get("itemsName")
965 if _items_name:
966 xml_name = _items_name
967 _items_ns = prop_meta.get("itemsNs")
968 if _items_ns is not None:
969 xml_ns = _items_ns
970 if xml_ns:
971 xml_name = "{" + xml_ns + "}" + xml_name
972 items = element.findall(xml_name) # pyright: ignore
973 if len(items) > 0:
974 existed_attr_keys.append(xml_name)
975 result[rf._rest_name] = _deserialize(rf._type, items)
976 elif not rf._is_optional:
977 existed_attr_keys.append(xml_name)
978 result[rf._rest_name] = []
979 continue
981 # text element is primitive type
982 if prop_meta.get("text", False):
983 if element.text is not None:
984 result[rf._rest_name] = _deserialize(rf._type, element.text)
985 continue
987 # wrapped element could be normal property or array
988 item = element.find(xml_name)
989 if item is not None:
990 existed_attr_keys.append(xml_name)
991 result[rf._rest_name] = _deserialize(rf._type, item)
993 # rest thing is additional properties
994 for e in element:
995 if e.tag not in existed_attr_keys:
996 result[e.tag] = _convert_element(e)
998 return result
1000 def copy(self) -> "Model":
1001 return Model(self.__dict__)
1003 def __new__(cls, *args: typing.Any, **kwargs: typing.Any) -> Self:
1004 if f"{cls.__module__}.{cls.__qualname__}" not in cls._calculated:
1005 # we know the last nine classes in mro are going to be 'Model', '_MyMutableMapping', 'MutableMapping',
1006 # 'Mapping', 'Collection', 'Sized', 'Iterable', 'Container' and 'object'
1007 mros = cls.__mro__[:-9][::-1] # ignore parents, and reverse the mro order
1008 attr_to_rest_field: dict[str, _RestField] = { # map attribute name to rest_field property
1009 k: v for mro_class in mros for k, v in mro_class.__dict__.items() if k[0] != "_" and hasattr(v, "_type")
1010 }
1011 annotations = {
1012 k: v
1013 for mro_class in mros
1014 if hasattr(mro_class, "__annotations__")
1015 for k, v in mro_class.__annotations__.items()
1016 }
1017 for attr, rf in attr_to_rest_field.items():
1018 rf._module = cls.__module__
1019 if not rf._type:
1020 rf._type = rf._get_deserialize_callable_from_annotation(annotations.get(attr, None))
1021 if not rf._rest_name_input:
1022 rf._rest_name_input = attr
1023 cls._attr_to_rest_field: dict[str, _RestField] = dict(attr_to_rest_field.items())
1024 # Build XML field plan for fast _init_from_xml (only for XML models)
1025 if getattr(cls, "_xml", None):
1026 cls._xml_field_plan = _build_xml_field_plan(cls, attr_to_rest_field)
1027 cls._calculated.add(f"{cls.__module__}.{cls.__qualname__}")
1029 return super().__new__(cls)
1031 def __init_subclass__(cls, discriminator: typing.Optional[str] = None) -> None:
1032 for base in cls.__bases__:
1033 if hasattr(base, "__mapping__"):
1034 base.__mapping__[discriminator or cls.__name__] = cls # type: ignore
1036 @classmethod
1037 def _get_discriminator(cls, exist_discriminators) -> typing.Optional["_RestField"]:
1038 for v in cls.__dict__.values():
1039 if isinstance(v, _RestField) and v._is_discriminator and v._rest_name not in exist_discriminators:
1040 return v
1041 return None
1043 @classmethod
1044 def _deserialize(cls, data, exist_discriminators):
1045 if not hasattr(cls, "__mapping__"):
1046 return cls(data)
1047 discriminator = cls._get_discriminator(exist_discriminators)
1048 if discriminator is None:
1049 return cls(data)
1050 exist_discriminators.append(discriminator._rest_name)
1051 if isinstance(data, ET.Element):
1052 model_meta = getattr(cls, "_xml", {})
1053 prop_meta = getattr(discriminator, "_xml", {})
1054 xml_name = prop_meta.get("name", discriminator._rest_name)
1055 xml_ns = _resolve_xml_ns(prop_meta, model_meta)
1056 if xml_ns:
1057 xml_name = "{" + xml_ns + "}" + xml_name
1059 if data.get(xml_name) is not None:
1060 discriminator_value = data.get(xml_name)
1061 else:
1062 discriminator_value = data.find(xml_name).text # pyright: ignore
1063 else:
1064 discriminator_value = data.get(discriminator._rest_name)
1065 mapped_cls = cls.__mapping__.get(discriminator_value, cls) # pyright: ignore # pylint: disable=no-member
1066 return mapped_cls._deserialize(data, exist_discriminators)
1068 def as_dict(self, *, exclude_readonly: bool = False) -> dict[str, typing.Any]:
1069 """Return a dict that can be turned into json using json.dump.
1071 :keyword bool exclude_readonly: Whether to remove the readonly properties.
1072 :returns: A dict JSON compatible object
1073 :rtype: dict
1074 """
1076 result = {}
1077 readonly_props = []
1078 if exclude_readonly:
1079 readonly_props = [p._rest_name for p in self._attr_to_rest_field.values() if _is_readonly(p)]
1080 for k, v in self.items():
1081 if exclude_readonly and k in readonly_props: # pyright: ignore
1082 continue
1083 is_multipart_file_input = False
1084 try:
1085 is_multipart_file_input = next(
1086 rf for rf in self._attr_to_rest_field.values() if rf._rest_name == k
1087 )._is_multipart_file_input
1088 except StopIteration:
1089 pass
1090 result[k] = v if is_multipart_file_input else Model._as_dict_value(v, exclude_readonly=exclude_readonly)
1091 return result
1093 @staticmethod
1094 def _as_dict_value(v: typing.Any, exclude_readonly: bool = False) -> typing.Any:
1095 if v is None or isinstance(v, _Null):
1096 return None
1097 if isinstance(v, (list, tuple, set)):
1098 return type(v)(Model._as_dict_value(x, exclude_readonly=exclude_readonly) for x in v)
1099 if isinstance(v, dict):
1100 return {dk: Model._as_dict_value(dv, exclude_readonly=exclude_readonly) for dk, dv in v.items()}
1101 return v.as_dict(exclude_readonly=exclude_readonly) if hasattr(v, "as_dict") else v
1104def _deserialize_model(model_deserializer: typing.Optional[typing.Callable], obj):
1105 if _is_model(obj):
1106 return obj
1107 return _deserialize(model_deserializer, obj)
1110def _deserialize_with_optional(if_obj_deserializer: typing.Optional[typing.Callable], obj):
1111 if obj is None:
1112 return obj
1113 return _deserialize_with_callable(if_obj_deserializer, obj)
1116def _deserialize_with_union(deserializers, obj):
1117 for deserializer in deserializers:
1118 try:
1119 return _deserialize(deserializer, obj)
1120 except DeserializationError:
1121 pass
1122 raise DeserializationError()
1125def _deserialize_dict(
1126 value_deserializer: typing.Optional[typing.Callable],
1127 module: typing.Optional[str],
1128 obj: dict[typing.Any, typing.Any],
1129):
1130 if obj is None:
1131 return obj
1132 if isinstance(obj, ET.Element):
1133 obj = {child.tag: child for child in obj}
1134 return {k: _deserialize(value_deserializer, v, module) for k, v in obj.items()}
1137def _deserialize_multiple_sequence(
1138 entry_deserializers: list[typing.Optional[typing.Callable]],
1139 module: typing.Optional[str],
1140 obj,
1141):
1142 if obj is None:
1143 return obj
1144 return type(obj)(_deserialize(deserializer, entry, module) for entry, deserializer in zip(obj, entry_deserializers))
1147def _is_array_encoded_deserializer(deserializer: functools.partial) -> bool:
1148 return (
1149 isinstance(deserializer, functools.partial)
1150 and isinstance(deserializer.args[0], functools.partial)
1151 and deserializer.args[0].func == _deserialize_array_encoded # pylint: disable=comparison-with-callable
1152 )
1155def _deserialize_sequence(
1156 deserializer: typing.Optional[typing.Callable],
1157 module: typing.Optional[str],
1158 obj,
1159):
1160 if obj is None:
1161 return obj
1162 if isinstance(obj, ET.Element):
1163 obj = list(obj)
1165 # encoded string may be deserialized to sequence
1166 if isinstance(obj, str) and isinstance(deserializer, functools.partial):
1167 # for list[str]
1168 if _is_array_encoded_deserializer(deserializer):
1169 return deserializer(obj)
1171 # for list[Union[...]]
1172 if isinstance(deserializer.args[0], list):
1173 for sub_deserializer in deserializer.args[0]:
1174 if _is_array_encoded_deserializer(sub_deserializer):
1175 return sub_deserializer(obj)
1177 return type(obj)(_deserialize(deserializer, entry, module) for entry in obj)
1180def _sorted_annotations(types: list[typing.Any]) -> list[typing.Any]:
1181 return sorted(
1182 types,
1183 key=lambda x: hasattr(x, "__name__") and x.__name__.lower() in ("str", "float", "int", "bool"),
1184 )
1187def _get_deserialize_callable_from_annotation( # pylint: disable=too-many-return-statements, too-many-statements, too-many-branches
1188 annotation: typing.Any,
1189 module: typing.Optional[str],
1190 rf: typing.Optional["_RestField"] = None,
1191) -> typing.Optional[typing.Callable[[typing.Any], typing.Any]]:
1192 if not annotation:
1193 return None
1195 # is it a type alias?
1196 if isinstance(annotation, str):
1197 if module is not None:
1198 annotation = _get_type_alias_type(module, annotation)
1200 # is it a forward ref / in quotes?
1201 if isinstance(annotation, (str, typing.ForwardRef)):
1202 try:
1203 model_name = annotation.__forward_arg__ # type: ignore
1204 except AttributeError:
1205 model_name = annotation
1206 if module is not None:
1207 annotation = _get_model(module, model_name) # type: ignore
1209 try:
1210 if module and _is_model(annotation):
1211 if rf:
1212 rf._is_model = True
1214 return functools.partial(_deserialize_model, annotation) # pyright: ignore
1215 except Exception:
1216 pass
1218 # is it a literal?
1219 try:
1220 if annotation.__origin__ is typing.Literal: # pyright: ignore
1221 return None
1222 except AttributeError:
1223 pass
1225 # is it optional?
1226 try:
1227 if any(a is _NONE_TYPE for a in annotation.__args__): # pyright: ignore
1228 if rf:
1229 rf._is_optional = True
1230 if len(annotation.__args__) <= 2: # pyright: ignore
1231 if_obj_deserializer = _get_deserialize_callable_from_annotation(
1232 next(a for a in annotation.__args__ if a is not _NONE_TYPE), module, rf # pyright: ignore
1233 )
1235 return functools.partial(_deserialize_with_optional, if_obj_deserializer)
1236 # the type is Optional[Union[...]], we need to remove the None type from the Union
1237 annotation_copy = copy.copy(annotation)
1238 annotation_copy.__args__ = [a for a in annotation_copy.__args__ if a is not _NONE_TYPE] # pyright: ignore
1239 return _get_deserialize_callable_from_annotation(annotation_copy, module, rf)
1240 except AttributeError:
1241 pass
1243 # is it union?
1244 if getattr(annotation, "__origin__", None) is typing.Union:
1245 # initial ordering is we make `string` the last deserialization option, because it is often them most generic
1246 deserializers = [
1247 _get_deserialize_callable_from_annotation(arg, module, rf)
1248 for arg in _sorted_annotations(annotation.__args__) # pyright: ignore
1249 ]
1251 return functools.partial(_deserialize_with_union, deserializers)
1253 try:
1254 annotation_name = (
1255 annotation.__name__ if hasattr(annotation, "__name__") else annotation._name # pyright: ignore
1256 )
1257 if annotation_name.lower() == "dict":
1258 value_deserializer = _get_deserialize_callable_from_annotation(
1259 annotation.__args__[1], module, rf # pyright: ignore
1260 )
1262 return functools.partial(
1263 _deserialize_dict,
1264 value_deserializer,
1265 module,
1266 )
1267 except (AttributeError, IndexError):
1268 pass
1269 try:
1270 annotation_name = (
1271 annotation.__name__ if hasattr(annotation, "__name__") else annotation._name # pyright: ignore
1272 )
1273 if annotation_name.lower() in ["list", "set", "tuple", "sequence"]:
1274 if len(annotation.__args__) > 1: # pyright: ignore
1275 entry_deserializers = [
1276 _get_deserialize_callable_from_annotation(dt, module, rf)
1277 for dt in annotation.__args__ # pyright: ignore
1278 ]
1279 return functools.partial(_deserialize_multiple_sequence, entry_deserializers, module)
1280 deserializer = _get_deserialize_callable_from_annotation(
1281 annotation.__args__[0], module, rf # pyright: ignore
1282 )
1284 return functools.partial(_deserialize_sequence, deserializer, module)
1285 except (TypeError, IndexError, AttributeError, SyntaxError):
1286 pass
1288 def _deserialize_default(
1289 deserializer,
1290 obj,
1291 ):
1292 if obj is None:
1293 return obj
1294 try:
1295 return _deserialize_with_callable(deserializer, obj)
1296 except Exception:
1297 pass
1298 return obj
1300 if get_deserializer(annotation, rf):
1301 return functools.partial(_deserialize_default, get_deserializer(annotation, rf))
1303 return functools.partial(_deserialize_default, annotation)
1306def _deserialize_with_callable(
1307 deserializer: typing.Optional[typing.Callable[[typing.Any], typing.Any]],
1308 value: typing.Any,
1309): # pylint: disable=too-many-return-statements
1310 try:
1311 if value is None or isinstance(value, _Null):
1312 return None
1313 if isinstance(value, ET.Element):
1314 if deserializer is str:
1315 return value.text or ""
1316 if deserializer is int:
1317 return int(value.text) if value.text else None
1318 if deserializer is float:
1319 return float(value.text) if value.text else None
1320 if deserializer is bool:
1321 return value.text == "true" if value.text else None
1322 if deserializer and deserializer in _DESERIALIZE_MAPPING.values():
1323 return deserializer(value.text) if value.text else None
1324 if deserializer and deserializer in _DESERIALIZE_MAPPING_WITHFORMAT.values():
1325 return deserializer(value.text) if value.text else None
1326 if deserializer is None:
1327 return value
1328 if deserializer in [int, float, bool]:
1329 return deserializer(value)
1330 if isinstance(deserializer, CaseInsensitiveEnumMeta):
1331 try:
1332 return deserializer(value.text if isinstance(value, ET.Element) else value)
1333 except ValueError:
1334 # for unknown value, return raw value
1335 return value.text if isinstance(value, ET.Element) else value
1336 if isinstance(deserializer, type) and issubclass(deserializer, Model):
1337 return deserializer._deserialize(value, [])
1338 return typing.cast(typing.Callable[[typing.Any], typing.Any], deserializer)(value)
1339 except Exception as e:
1340 raise DeserializationError() from e
1343def _deserialize(
1344 deserializer: typing.Any,
1345 value: typing.Any,
1346 module: typing.Optional[str] = None,
1347 rf: typing.Optional["_RestField"] = None,
1348 format: typing.Optional[str] = None,
1349) -> typing.Any:
1350 if isinstance(value, PipelineResponse):
1351 value = value.http_response.json()
1352 if rf is None and format:
1353 rf = _RestField(format=format)
1354 if not isinstance(deserializer, functools.partial):
1355 deserializer = _get_deserialize_callable_from_annotation(deserializer, module, rf)
1356 return _deserialize_with_callable(deserializer, value)
1359def _failsafe_deserialize(
1360 deserializer: typing.Any,
1361 response: HttpResponse,
1362 module: typing.Optional[str] = None,
1363 rf: typing.Optional["_RestField"] = None,
1364 format: typing.Optional[str] = None,
1365) -> typing.Any:
1366 try:
1367 return _deserialize(deserializer, response.json(), module, rf, format)
1368 except Exception: # pylint: disable=broad-except
1369 _LOGGER.warning(
1370 "Ran into a deserialization error. Ignoring since this is failsafe deserialization", exc_info=True
1371 )
1372 return None
1375def _failsafe_deserialize_xml(
1376 deserializer: typing.Any,
1377 response: HttpResponse,
1378) -> typing.Any:
1379 try:
1380 return _deserialize_xml(deserializer, response.text())
1381 except Exception: # pylint: disable=broad-except
1382 _LOGGER.warning(
1383 "Ran into a deserialization error. Ignoring since this is failsafe deserialization", exc_info=True
1384 )
1385 return None
1388# pylint: disable=too-many-instance-attributes
1389class _RestField:
1390 def __init__(
1391 self,
1392 *,
1393 name: typing.Optional[str] = None,
1394 type: typing.Optional[typing.Callable] = None, # pylint: disable=redefined-builtin
1395 is_discriminator: bool = False,
1396 visibility: typing.Optional[list[str]] = None,
1397 default: typing.Any = _UNSET,
1398 format: typing.Optional[str] = None,
1399 is_multipart_file_input: bool = False,
1400 xml: typing.Optional[dict[str, typing.Any]] = None,
1401 deserializer: typing.Optional[typing.Callable] = None,
1402 ):
1403 self._type = type
1404 self._rest_name_input = name
1405 self._module: typing.Optional[str] = None
1406 self._is_discriminator = is_discriminator
1407 self._visibility = visibility
1408 self._is_model = False
1409 self._is_optional = False
1410 self._default = default
1411 self._format = format
1412 self._is_multipart_file_input = is_multipart_file_input
1413 self._xml = xml if xml is not None else {}
1414 self._deserializer = deserializer
1416 @property
1417 def _class_type(self) -> typing.Any:
1418 result = getattr(self._type, "args", [None])[0]
1419 # type may be wrapped by nested functools.partial so we need to check for that
1420 if isinstance(result, functools.partial):
1421 return getattr(result, "args", [None])[0]
1422 return result
1424 @property
1425 def _rest_name(self) -> str:
1426 if self._rest_name_input is None:
1427 raise ValueError("Rest name was never set")
1428 return self._rest_name_input
1430 def __get__(self, obj: Model, type=None): # pylint: disable=redefined-builtin
1431 # by this point, type and rest_name will have a value bc we default
1432 # them in __new__ of the Model class
1433 # Use _data.get() directly to avoid triggering __getitem__ which clears the cache
1434 item = obj._data.get(self._rest_name, _UNSET)
1435 if item is _UNSET:
1436 # Field not set by user; return the client default if one exists, otherwise None
1437 return self._default if self._default is not _UNSET else None
1438 if item is None:
1439 return item
1440 if self._is_model:
1441 return item
1443 # For mutable types, we want mutations to directly affect _data
1444 # Check if we've already deserialized this value
1445 cache_attr = f"_deserialized_{self._rest_name}"
1446 if hasattr(obj, cache_attr):
1447 # Return the value from _data directly (it's been deserialized in place)
1448 return obj._data.get(self._rest_name)
1450 # Fast path: use _deserializer directly (avoids _serialize/_deserialize chain)
1451 if self._deserializer:
1452 deserialized = self._deserializer(item)
1453 else:
1454 deserialized = _deserialize(self._type, _serialize(item, self._format), rf=self)
1456 # For mutable types, store the deserialized value back in _data
1457 # so mutations directly affect _data
1458 if isinstance(deserialized, (dict, list, set)):
1459 obj._data[self._rest_name] = deserialized
1460 object.__setattr__(obj, cache_attr, True) # Mark as deserialized
1461 return deserialized
1463 return deserialized
1465 def __set__(self, obj: Model, value) -> None:
1466 # Clear the cached deserialized object when setting a new value
1467 cache_attr = f"_deserialized_{self._rest_name}"
1468 if hasattr(obj, cache_attr):
1469 object.__delattr__(obj, cache_attr)
1471 if value is None:
1472 # we want to wipe out entries if users set attr to None
1473 try:
1474 obj.__delitem__(self._rest_name)
1475 except KeyError:
1476 pass
1477 return
1478 if self._is_model:
1479 if not _is_model(value):
1480 value = _deserialize(self._type, value)
1481 obj.__setitem__(self._rest_name, value)
1482 return
1483 obj.__setitem__(self._rest_name, _serialize(value, self._format))
1485 def _get_deserialize_callable_from_annotation(
1486 self, annotation: typing.Any
1487 ) -> typing.Optional[typing.Callable[[typing.Any], typing.Any]]:
1488 return _get_deserialize_callable_from_annotation(annotation, self._module, self)
1491def rest_field(
1492 *,
1493 name: typing.Optional[str] = None,
1494 type: typing.Optional[typing.Callable] = None, # pylint: disable=redefined-builtin
1495 visibility: typing.Optional[list[str]] = None,
1496 default: typing.Any = _UNSET,
1497 format: typing.Optional[str] = None,
1498 is_multipart_file_input: bool = False,
1499 xml: typing.Optional[dict[str, typing.Any]] = None,
1500 deserializer: typing.Optional[typing.Callable] = None,
1501) -> typing.Any:
1502 return _RestField(
1503 name=name,
1504 type=type,
1505 visibility=visibility,
1506 default=default,
1507 format=format,
1508 is_multipart_file_input=is_multipart_file_input,
1509 xml=xml,
1510 deserializer=deserializer,
1511 )
1514def rest_discriminator(
1515 *,
1516 name: typing.Optional[str] = None,
1517 type: typing.Optional[typing.Callable] = None, # pylint: disable=redefined-builtin
1518 visibility: typing.Optional[list[str]] = None,
1519 xml: typing.Optional[dict[str, typing.Any]] = None,
1520) -> typing.Any:
1521 return _RestField(name=name, type=type, is_discriminator=True, visibility=visibility, xml=xml)
1524def serialize_xml(model: Model, exclude_readonly: bool = False) -> str:
1525 """Serialize a model to XML.
1527 :param Model model: The model to serialize.
1528 :param bool exclude_readonly: Whether to exclude readonly properties.
1529 :returns: The XML representation of the model.
1530 :rtype: str
1531 """
1532 return ET.tostring(_get_element(model, exclude_readonly), encoding="unicode") # type: ignore
1535def _get_xml_ns(meta: dict[str, typing.Any]) -> typing.Optional[str]:
1536 """Return the XML namespace from a metadata dict, checking both 'ns' (old-style) and 'namespace' (DPG) keys.
1538 :param dict meta: The metadata dictionary to extract namespace from.
1539 :returns: The namespace string if 'ns' or 'namespace' key is present, None otherwise.
1540 :rtype: str or None
1541 """
1542 ns = meta.get("ns")
1543 if ns is None:
1544 ns = meta.get("namespace")
1545 return ns
1548def _resolve_xml_ns(
1549 prop_meta: dict[str, typing.Any], model_meta: typing.Optional[dict[str, typing.Any]] = None
1550) -> typing.Optional[str]:
1551 """Resolve XML namespace for a property, falling back to model namespace when appropriate.
1553 Checks the property metadata first; if no namespace is found and the model does not declare
1554 an explicit prefix, falls back to the model-level namespace.
1556 :param dict prop_meta: The property metadata dictionary.
1557 :param dict model_meta: The model metadata dictionary, used as fallback.
1558 :returns: The resolved namespace string, or None.
1559 :rtype: str or None
1560 """
1561 ns = _get_xml_ns(prop_meta)
1562 if ns is None and model_meta is not None and not model_meta.get("prefix"):
1563 ns = _get_xml_ns(model_meta)
1564 return ns
1567def _set_xml_attribute(element: ET.Element, name: str, value: typing.Any, prop_meta: dict[str, typing.Any]) -> None:
1568 """Set an XML attribute on an element, handling namespace prefix registration.
1570 :param ET.Element element: The element to set the attribute on.
1571 :param str name: The default attribute name (wire name).
1572 :param any value: The attribute value.
1573 :param dict prop_meta: The property metadata dictionary.
1574 """
1575 xml_name = prop_meta.get("name", name)
1576 _attr_ns = _get_xml_ns(prop_meta)
1577 if _attr_ns:
1578 _attr_prefix = prop_meta.get("prefix")
1579 if _attr_prefix:
1580 _safe_register_namespace(_attr_prefix, _attr_ns)
1581 xml_name = "{" + _attr_ns + "}" + xml_name
1582 element.set(xml_name, _get_primitive_type_value(value))
1585def _get_element(
1586 o: typing.Any,
1587 exclude_readonly: bool = False,
1588 parent_meta: typing.Optional[dict[str, typing.Any]] = None,
1589 wrapped_element: typing.Optional[ET.Element] = None,
1590) -> typing.Union[ET.Element, list[ET.Element]]:
1591 if _is_model(o):
1592 model_meta = getattr(o, "_xml", {})
1594 # if prop is a model, then use the prop element directly, else generate a wrapper of model
1595 if wrapped_element is None:
1596 # When serializing as an array item (parent_meta is set), check if the parent has an
1597 # explicit itemsName. This ensures correct element names for unwrapped arrays (where
1598 # the element tag is the property/items name, not the model type name).
1599 _items_name = parent_meta.get("itemsName") if parent_meta is not None else None
1600 element_name = _items_name if _items_name else (model_meta.get("name") or o.__class__.__name__)
1601 _model_ns = _get_xml_ns(model_meta)
1602 wrapped_element = _create_xml_element(
1603 element_name,
1604 model_meta.get("prefix"),
1605 _model_ns,
1606 )
1608 readonly_props = []
1609 if exclude_readonly:
1610 readonly_props = [p._rest_name for p in o._attr_to_rest_field.values() if _is_readonly(p)]
1612 for k, v in o.items():
1613 # do not serialize readonly properties
1614 if exclude_readonly and k in readonly_props:
1615 continue
1617 prop_rest_field = _get_rest_field(o._attr_to_rest_field, k)
1618 if prop_rest_field:
1619 prop_meta = getattr(prop_rest_field, "_xml").copy()
1620 # use the wire name as xml name if no specific name is set
1621 if prop_meta.get("name") is None:
1622 prop_meta["name"] = k
1623 else:
1624 # additional properties will not have rest field, use the wire name as xml name
1625 prop_meta = {"name": k}
1627 # Propagate model namespace to properties only for old-style "ns"-keyed models.
1628 # DPG-generated models use the "namespace" key and explicitly declare namespace on
1629 # each property that needs it, so propagation is intentionally skipped for them.
1630 if prop_meta.get("ns") is None and model_meta.get("ns"):
1631 prop_meta["ns"] = model_meta.get("ns")
1632 prop_meta["prefix"] = model_meta.get("prefix")
1634 if prop_meta.get("unwrapped", False):
1635 # unwrapped could only set on array
1636 wrapped_element.extend(_get_element(v, exclude_readonly, prop_meta))
1637 elif prop_meta.get("text", False):
1638 # text could only set on primitive type
1639 wrapped_element.text = _get_primitive_type_value(v)
1640 elif prop_meta.get("attribute", False):
1641 _set_xml_attribute(wrapped_element, k, v, prop_meta)
1642 else:
1643 # other wrapped prop element
1644 wrapped_element.append(_get_wrapped_element(v, exclude_readonly, prop_meta))
1645 return wrapped_element
1646 if isinstance(o, list):
1647 return [_get_element(x, exclude_readonly, parent_meta) for x in o] # type: ignore
1648 if isinstance(o, dict):
1649 result = []
1650 _dict_ns = _get_xml_ns(parent_meta) if parent_meta else None
1651 for k, v in o.items():
1652 result.append(
1653 _get_wrapped_element(
1654 v,
1655 exclude_readonly,
1656 {
1657 "name": k,
1658 "ns": _dict_ns,
1659 "prefix": parent_meta.get("prefix") if parent_meta else None,
1660 },
1661 )
1662 )
1663 return result
1665 # primitive case need to create element based on parent_meta
1666 if parent_meta:
1667 _items_ns = parent_meta.get("itemsNs")
1668 if _items_ns is None:
1669 _items_ns = _get_xml_ns(parent_meta)
1670 return _get_wrapped_element(
1671 o,
1672 exclude_readonly,
1673 {
1674 "name": parent_meta.get("itemsName", parent_meta.get("name")),
1675 "prefix": parent_meta.get("itemsPrefix", parent_meta.get("prefix")),
1676 "ns": _items_ns,
1677 },
1678 )
1680 raise ValueError("Could not serialize value into xml: " + o)
1683def _get_wrapped_element(
1684 v: typing.Any,
1685 exclude_readonly: bool,
1686 meta: typing.Optional[dict[str, typing.Any]],
1687) -> ET.Element:
1688 _meta_ns = _get_xml_ns(meta) if meta else None
1689 wrapped_element = _create_xml_element(
1690 meta.get("name") if meta else None, meta.get("prefix") if meta else None, _meta_ns
1691 )
1692 if isinstance(v, (dict, list)):
1693 wrapped_element.extend(_get_element(v, exclude_readonly, meta))
1694 elif _is_model(v):
1695 _get_element(v, exclude_readonly, meta, wrapped_element)
1696 else:
1697 wrapped_element.text = _get_primitive_type_value(v)
1698 return wrapped_element # type: ignore[no-any-return]
1701def _get_primitive_type_value(v) -> str:
1702 if v is True:
1703 return "true"
1704 if v is False:
1705 return "false"
1706 if isinstance(v, _Null):
1707 return ""
1708 return str(v)
1711def _safe_register_namespace(prefix: str, ns: str) -> None:
1712 """Register an XML namespace prefix, handling reserved prefix patterns.
1714 Some prefixes (e.g. 'ns2') match Python's reserved 'ns\\d+' pattern used for
1715 auto-generated prefixes, causing register_namespace to raise ValueError.
1716 Falls back to directly registering in the internal namespace map.
1718 :param str prefix: The namespace prefix to register.
1719 :param str ns: The namespace URI.
1720 """
1721 try:
1722 ET.register_namespace(prefix, ns)
1723 except ValueError:
1724 _ns_map = getattr(ET, "_namespace_map", None)
1725 if _ns_map is not None:
1726 _ns_map[ns] = prefix
1729def _create_xml_element(
1730 tag: typing.Any, prefix: typing.Optional[str] = None, ns: typing.Optional[str] = None
1731) -> ET.Element:
1732 if prefix and ns:
1733 _safe_register_namespace(prefix, ns)
1734 if ns:
1735 return ET.Element("{" + ns + "}" + tag)
1736 return ET.Element(tag)
1739def _deserialize_xml(
1740 deserializer: typing.Any,
1741 value: str,
1742) -> typing.Any:
1743 element = ET.fromstring(value) # nosec
1744 if _is_model(deserializer):
1745 return deserializer._deserialize(element, [])
1746 return _deserialize(deserializer, element)
1749def _convert_element(e: ET.Element):
1750 # dict case
1751 if len(e.attrib) > 0 or len({child.tag for child in e}) > 1:
1752 dict_result: dict[str, typing.Any] = {}
1753 for child in e:
1754 if dict_result.get(child.tag) is not None:
1755 if isinstance(dict_result[child.tag], list):
1756 dict_result[child.tag].append(_convert_element(child))
1757 else:
1758 dict_result[child.tag] = [dict_result[child.tag], _convert_element(child)]
1759 else:
1760 dict_result[child.tag] = _convert_element(child)
1761 dict_result.update(e.attrib)
1762 return dict_result
1763 # array case
1764 if len(e) > 0:
1765 array_result: list[typing.Any] = []
1766 for child in e:
1767 array_result.append(_convert_element(child))
1768 return array_result
1769 # primitive case
1770 return e.text