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 the mapping'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 the mapping's values
469 :rtype: ~typing.ValuesView
470 """
471 return self._data.values()
473 def items(self) -> typing.ItemsView[str, typing.Any]:
474 """
475 :returns: a set-like object providing a view on the mapping'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: The value for key if key is in the dictionary, else default.
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 the dictionary is empty.
521 """
522 return self._data.popitem()
524 def clear(self) -> None:
525 """
526 Remove all items from the dictionary.
527 """
528 self._data.clear()
530 def update(self, *args: typing.Any, **kwargs: typing.Any) -> None: # pylint: disable=arguments-differ
531 """
532 Update the dictionary from a mapping or an iterable of key-value pairs.
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 Return the value for key if key is in the dictionary; otherwise set the key to
546 default and return default.
547 :param str key: The key to look up.
548 :param any default: The value to set if key is not in the dictionary
549 :returns: The value for key if key is in the dictionary, else default.
550 :rtype: any
551 """
552 if default is _UNSET:
553 return self._data.setdefault(key)
554 return self._data.setdefault(key, default)
556 def __eq__(self, other: typing.Any) -> bool:
557 if isinstance(other, _MyMutableMapping):
558 return self._data == other._data
559 try:
560 other_model = self.__class__(other)
561 except Exception:
562 return False
563 return self._data == other_model._data
565 def __repr__(self) -> str:
566 return str(self._data)
569def _is_model(obj: typing.Any) -> bool:
570 return getattr(obj, "_is_model", False)
573def _serialize(o, format: typing.Optional[str] = None): # pylint: disable=too-many-return-statements
574 if isinstance(o, list):
575 if format in _ARRAY_ENCODE_MAPPING and all(isinstance(x, str) for x in o):
576 return _ARRAY_ENCODE_MAPPING[format].join(o)
577 return [_serialize(x, format) for x in o]
578 if isinstance(o, dict):
579 return {k: _serialize(v, format) for k, v in o.items()}
580 if isinstance(o, set):
581 return {_serialize(x, format) for x in o}
582 if isinstance(o, tuple):
583 return tuple(_serialize(x, format) for x in o)
584 if isinstance(o, (bytes, bytearray)):
585 return _serialize_bytes(o, format)
586 if isinstance(o, decimal.Decimal):
587 return float(o)
588 if isinstance(o, enum.Enum):
589 return o.value
590 if isinstance(o, int):
591 if format == "str":
592 return str(o)
593 return o
594 try:
595 # First try datetime.datetime
596 return _serialize_datetime(o, format)
597 except AttributeError:
598 pass
599 # Last, try datetime.timedelta
600 try:
601 return _serialize_duration(o, format)
602 except AttributeError:
603 # This will be raised when it hits value.total_seconds in the method above
604 pass
605 return o
608def _get_rest_field(attr_to_rest_field: dict[str, "_RestField"], rest_name: str) -> typing.Optional["_RestField"]:
609 try:
610 return next(rf for rf in attr_to_rest_field.values() if rf._rest_name == rest_name)
611 except StopIteration:
612 return None
615def _create_value(rf: typing.Optional["_RestField"], value: typing.Any) -> typing.Any:
616 if not rf:
617 return _serialize(value, None)
618 if rf._is_multipart_file_input:
619 return value
620 if rf._is_model:
621 return _deserialize(rf._type, value)
622 if isinstance(value, ET.Element):
623 value = _deserialize(rf._type, value)
624 return _serialize(value, rf._format)
627# ============================================================================
628# Fast-path scalar deserializer functions for rest_field(deserializer=...)
629# These are referenced from rest_field declarations to bypass the generic
630# _deserialize -> _deserialize_with_callable chain.
631# Only simple/primitive types — no models or container types.
632# ============================================================================
635def _xml_deser_str(value):
636 if isinstance(value, ET.Element):
637 return value.text or ""
638 return str(value) if value is not None else None
641def _xml_deser_int(value):
642 if isinstance(value, ET.Element):
643 return int(value.text) if value.text else None
644 return int(value) if value is not None else None
647def _xml_deser_float(value):
648 if isinstance(value, ET.Element):
649 return float(value.text) if value.text else None
650 return float(value) if value is not None else None
653def _xml_deser_bool(value):
654 if isinstance(value, ET.Element):
655 text = value.text
656 else:
657 text = value
658 if text is None:
659 return None
660 if text in (True, False):
661 return text
662 return text.lower() == "true"
665# pylint: disable=docstring-missing-param
666def _xml_deser_bytes(value):
667 """Deserialize bytes from XML (base64)."""
668 if isinstance(value, ET.Element):
669 text = value.text
670 else:
671 text = value
672 if text is None:
673 return None
674 return _deserialize_bytes(text)
677def _xml_deser_bytes_base64url(value):
678 """Deserialize bytes from XML (base64url)."""
679 if isinstance(value, ET.Element):
680 text = value.text
681 else:
682 text = value
683 if text is None:
684 return None
685 return _deserialize_bytes_base64(text)
688def _xml_deser_datetime(value):
689 """Deserialize a datetime from XML (ISO 8601 / rfc3339)."""
690 if isinstance(value, ET.Element):
691 text = value.text
692 else:
693 text = value
694 if text is None:
695 return None
696 return _deserialize_datetime(text)
699def _xml_deser_datetime_rfc7231(value):
700 """Deserialize a datetime from XML (RFC7231 format)."""
701 if isinstance(value, ET.Element):
702 text = value.text
703 else:
704 text = value
705 if text is None:
706 return None
707 return _deserialize_datetime_rfc7231(text)
710def _xml_deser_datetime_unix_timestamp(value):
711 """Deserialize a datetime from XML (Unix timestamp)."""
712 if isinstance(value, ET.Element):
713 text = value.text
714 else:
715 text = value
716 if text is None:
717 return None
718 return _deserialize_datetime_unix_timestamp(float(text))
721def _xml_deser_date(value):
722 """Deserialize a date from XML (ISO 8601)."""
723 if isinstance(value, ET.Element):
724 text = value.text
725 else:
726 text = value
727 if text is None:
728 return None
729 return _deserialize_date(text)
732def _xml_deser_time(value):
733 """Deserialize a time from XML (ISO 8601)."""
734 if isinstance(value, ET.Element):
735 text = value.text
736 else:
737 text = value
738 if text is None:
739 return None
740 return _deserialize_time(text)
743def _xml_deser_duration(value):
744 """Deserialize a timedelta from XML (ISO 8601 duration)."""
745 if isinstance(value, ET.Element):
746 text = value.text
747 else:
748 text = value
749 if text is None:
750 return None
751 return _deserialize_duration(text)
754def _xml_deser_decimal(value):
755 """Deserialize a Decimal from XML."""
756 if isinstance(value, ET.Element):
757 text = value.text
758 else:
759 text = value
760 if text is None:
761 return None
762 return _deserialize_decimal(text)
765def _xml_deser_enum_or_str(enum_cls, value):
766 """Deserialize a Union[EnumType, str] from XML."""
767 text = value.text if isinstance(value, ET.Element) else value
768 if text is None:
769 return None
770 try:
771 return enum_cls(text)
772 except ValueError:
773 return text
776def _extract_xml_model_type(rf_type):
777 """Extract the concrete Model class from a resolved rf._type partial chain.
779 Unwraps ``Optional[Model]`` and ``_deserialize_model(Model, ...)``
780 wrappers. Only handles Model and Optional[Model] — other composite
781 types (List, Dict, Union, etc.) return None and fall through to the
782 generic ``_deserialize`` path at runtime.
783 """
784 if rf_type is None:
785 return None
786 if isinstance(rf_type, type) and _is_model(rf_type):
787 return rf_type
788 if not isinstance(rf_type, functools.partial):
789 return None
790 func = rf_type.func
791 args = rf_type.args
792 if func is _deserialize_with_optional and args:
793 return _extract_xml_model_type(args[0])
794 if func is _deserialize_model and args:
795 cls = args[0]
796 return cls if isinstance(cls, type) and _is_model(cls) else None
797 return None
800def _build_xml_field_plan( # pylint: disable=docstring-missing-return, docstring-missing-rtype, unused-variable
801 cls, attr_to_rest_field: dict
802) -> list:
803 """Build a precomputed XML field plan for fast _init_from_xml iteration.
805 Called once per model class in __new__. Returns a list of tuples:
806 (rest_name, xml_name, kind, deser, rf_type, is_optional, items_name)
808 kind: 0=wrapped, 1=attribute, 2=unwrapped, 3=text
810 For Model and Optional[Model] fields that lack a scalar
811 ``_deserializer``, this function precomputes the Model class as the
812 deserializer so ``_init_from_xml`` can call ``ModelClass(element)``
813 directly instead of going through the expensive
814 ``_get_deserialize_callable_from_annotation`` chain at runtime.
815 """
816 model_meta = getattr(cls, "_xml", {})
817 model_ns = model_meta.get("ns") or model_meta.get("namespace")
818 plan = []
820 for rf in attr_to_rest_field.values():
821 prop_meta = getattr(rf, "_xml", {})
822 deser = rf._deserializer
824 xml_name = prop_meta.get("name", rf._rest_name)
825 xml_ns = _resolve_xml_ns(prop_meta, model_meta)
826 if xml_ns:
827 xml_name = "{" + xml_ns + "}" + xml_name
829 is_optional = rf._is_optional
831 # For Model / Optional[Model] fields without a scalar deserializer,
832 # precompute the Model class as the deserializer.
833 if deser is None and rf._type is not None:
834 model_cls = _extract_xml_model_type(rf._type)
835 if model_cls is not None:
836 deser = model_cls
838 if prop_meta.get("attribute", False):
839 plan.append((rf._rest_name, xml_name, 1, deser, rf._type, is_optional, None))
840 elif prop_meta.get("unwrapped", False):
841 items_name = prop_meta.get("itemsName")
842 if items_name:
843 items_ns = prop_meta.get("itemsNs")
844 if items_ns is not None:
845 xml_ns = items_ns
846 if xml_ns:
847 items_name = "{" + xml_ns + "}" + items_name
848 else:
849 items_name = xml_name
850 plan.append((rf._rest_name, xml_name, 2, deser, rf._type, is_optional, items_name))
851 elif prop_meta.get("text", False):
852 plan.append((rf._rest_name, xml_name, 3, deser, rf._type, is_optional, None))
853 else:
854 plan.append((rf._rest_name, xml_name, 0, deser, rf._type, is_optional, None))
856 return plan
859# pylint: enable=docstring-missing-param
860class Model(_MyMutableMapping):
861 _is_model = True
862 # label whether current class's _attr_to_rest_field has been calculated
863 # could not see _attr_to_rest_field directly because subclass inherits it from parent class
864 _calculated: set[str] = set()
866 def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
867 class_name = self.__class__.__name__
868 if len(args) > 1:
869 raise TypeError(f"{class_name}.__init__() takes 2 positional arguments but {len(args) + 1} were given")
870 dict_to_pass: dict[str, typing.Any] = {}
871 if args:
872 if isinstance(args[0], ET.Element):
873 dict_to_pass.update(self._init_from_xml(args[0]))
874 else:
875 dict_to_pass.update(
876 {k: _create_value(_get_rest_field(self._attr_to_rest_field, k), v) for k, v in args[0].items()}
877 )
878 else:
879 non_attr_kwargs = [k for k in kwargs if k not in self._attr_to_rest_field]
880 if non_attr_kwargs:
881 # actual type errors only throw the first wrong keyword arg they see, so following that.
882 raise TypeError(f"{class_name}.__init__() got an unexpected keyword argument '{non_attr_kwargs[0]}'")
883 dict_to_pass.update(
884 {
885 self._attr_to_rest_field[k]._rest_name: _create_value(self._attr_to_rest_field[k], v)
886 for k, v in kwargs.items()
887 if v is not None
888 }
889 )
890 # Apply client default values for fields the caller didn't set so that
891 # defaults are part of `_data` and therefore included during serialization.
892 for rf in self._attr_to_rest_field.values():
893 if rf._default is _UNSET:
894 continue
895 if rf._rest_name in dict_to_pass:
896 continue
897 dict_to_pass[rf._rest_name] = _create_value(rf, rf._default)
898 super().__init__(dict_to_pass)
900 def _init_from_xml( # pylint: disable=too-many-branches, too-many-statements
901 self, element: ET.Element
902 ) -> dict[str, typing.Any]:
903 """Deserialize an XML element into a dict mapping rest field names to values.
905 :param ET.Element element: The XML element to deserialize from.
906 :returns: A dictionary of rest_name to deserialized value pairs.
907 :rtype: dict
908 """
909 result: dict[str, typing.Any] = {}
910 existed_attr_keys: list[str] = []
912 field_plan = getattr(self, "_xml_field_plan", None)
913 if field_plan:
914 for rest_name, xml_name, kind, deser, rf_type, is_optional, items_name in field_plan:
915 if kind == 0: # wrapped element (most common)
916 item = element.find(xml_name)
917 if item is not None:
918 existed_attr_keys.append(xml_name)
919 if deser:
920 result[rest_name] = deser(item)
921 else:
922 result[rest_name] = _deserialize(rf_type, item)
923 elif kind == 1: # attribute
924 attr_val = element.get(xml_name)
925 if attr_val is not None:
926 existed_attr_keys.append(xml_name)
927 if deser:
928 result[rest_name] = deser(attr_val)
929 else:
930 result[rest_name] = attr_val
931 elif kind == 2: # unwrapped array
932 items = element.findall(items_name) # pyright: ignore
933 if len(items) > 0:
934 existed_attr_keys.append(items_name)
935 if deser:
936 result[rest_name] = deser(items)
937 else:
938 result[rest_name] = _deserialize(rf_type, items)
939 elif not is_optional:
940 existed_attr_keys.append(items_name)
941 result[rest_name] = []
942 elif kind == 3: # text
943 if element.text is not None:
944 if deser:
945 result[rest_name] = deser(element.text)
946 else:
947 result[rest_name] = element.text
948 else:
949 model_meta = getattr(self, "_xml", {})
950 for rf in self._attr_to_rest_field.values():
951 prop_meta = getattr(rf, "_xml", {})
952 xml_name = prop_meta.get("name", rf._rest_name)
953 xml_ns = _resolve_xml_ns(prop_meta, model_meta)
954 if xml_ns:
955 xml_name = "{" + xml_ns + "}" + xml_name
957 # attribute
958 if prop_meta.get("attribute", False) and element.get(xml_name) is not None:
959 existed_attr_keys.append(xml_name)
960 result[rf._rest_name] = _deserialize(rf._type, element.get(xml_name))
961 continue
963 # unwrapped element is array
964 if prop_meta.get("unwrapped", False):
965 _items_name = prop_meta.get("itemsName")
966 if _items_name:
967 xml_name = _items_name
968 _items_ns = prop_meta.get("itemsNs")
969 if _items_ns is not None:
970 xml_ns = _items_ns
971 if xml_ns:
972 xml_name = "{" + xml_ns + "}" + xml_name
973 items = element.findall(xml_name) # pyright: ignore
974 if len(items) > 0:
975 existed_attr_keys.append(xml_name)
976 result[rf._rest_name] = _deserialize(rf._type, items)
977 elif not rf._is_optional:
978 existed_attr_keys.append(xml_name)
979 result[rf._rest_name] = []
980 continue
982 # text element is primitive type
983 if prop_meta.get("text", False):
984 if element.text is not None:
985 result[rf._rest_name] = _deserialize(rf._type, element.text)
986 continue
988 # wrapped element could be normal property or array
989 item = element.find(xml_name)
990 if item is not None:
991 existed_attr_keys.append(xml_name)
992 result[rf._rest_name] = _deserialize(rf._type, item)
994 # rest thing is additional properties
995 for e in element:
996 if e.tag not in existed_attr_keys:
997 result[e.tag] = _convert_element(e)
999 return result
1001 def copy(self) -> "Model":
1002 return Model(self.__dict__)
1004 def __new__(cls, *args: typing.Any, **kwargs: typing.Any) -> Self:
1005 if f"{cls.__module__}.{cls.__qualname__}" not in cls._calculated:
1006 # we know the last nine classes in mro are going to be 'Model', '_MyMutableMapping', 'MutableMapping',
1007 # 'Mapping', 'Collection', 'Sized', 'Iterable', 'Container' and 'object'
1008 mros = cls.__mro__[:-9][::-1] # ignore parents, and reverse the mro order
1009 attr_to_rest_field: dict[str, _RestField] = { # map attribute name to rest_field property
1010 k: v for mro_class in mros for k, v in mro_class.__dict__.items() if k[0] != "_" and hasattr(v, "_type")
1011 }
1012 annotations = {
1013 k: v
1014 for mro_class in mros
1015 if hasattr(mro_class, "__annotations__")
1016 for k, v in mro_class.__annotations__.items()
1017 }
1018 for attr, rf in attr_to_rest_field.items():
1019 rf._module = cls.__module__
1020 if not rf._type:
1021 rf._type = rf._get_deserialize_callable_from_annotation(annotations.get(attr, None))
1022 if not rf._rest_name_input:
1023 rf._rest_name_input = attr
1024 cls._attr_to_rest_field: dict[str, _RestField] = dict(attr_to_rest_field.items())
1025 # Build XML field plan for fast _init_from_xml (only for XML models)
1026 if getattr(cls, "_xml", None):
1027 cls._xml_field_plan = _build_xml_field_plan(cls, attr_to_rest_field)
1028 cls._calculated.add(f"{cls.__module__}.{cls.__qualname__}")
1030 return super().__new__(cls)
1032 def __init_subclass__(cls, discriminator: typing.Optional[str] = None) -> None:
1033 for base in cls.__bases__:
1034 if hasattr(base, "__mapping__"):
1035 base.__mapping__[discriminator or cls.__name__] = cls # type: ignore
1037 @classmethod
1038 def _get_discriminator(cls, exist_discriminators) -> typing.Optional["_RestField"]:
1039 for v in cls.__dict__.values():
1040 if isinstance(v, _RestField) and v._is_discriminator and v._rest_name not in exist_discriminators:
1041 return v
1042 return None
1044 @classmethod
1045 def _deserialize(cls, data, exist_discriminators):
1046 if not hasattr(cls, "__mapping__"):
1047 return cls(data)
1048 discriminator = cls._get_discriminator(exist_discriminators)
1049 if discriminator is None:
1050 return cls(data)
1051 exist_discriminators.append(discriminator._rest_name)
1052 if isinstance(data, ET.Element):
1053 model_meta = getattr(cls, "_xml", {})
1054 prop_meta = getattr(discriminator, "_xml", {})
1055 xml_name = prop_meta.get("name", discriminator._rest_name)
1056 xml_ns = _resolve_xml_ns(prop_meta, model_meta)
1057 if xml_ns:
1058 xml_name = "{" + xml_ns + "}" + xml_name
1060 if data.get(xml_name) is not None:
1061 discriminator_value = data.get(xml_name)
1062 else:
1063 discriminator_value = data.find(xml_name).text # pyright: ignore
1064 else:
1065 discriminator_value = data.get(discriminator._rest_name)
1066 mapped_cls = cls.__mapping__.get(discriminator_value, cls) # pyright: ignore # pylint: disable=no-member
1067 return mapped_cls._deserialize(data, exist_discriminators)
1069 def as_dict(self, *, exclude_readonly: bool = False) -> dict[str, typing.Any]:
1070 """Return a dict that can be turned into json using json.dump.
1072 :keyword bool exclude_readonly: Whether to remove the readonly properties.
1073 :returns: A dict JSON compatible object
1074 :rtype: dict
1075 """
1077 result = {}
1078 readonly_props = []
1079 if exclude_readonly:
1080 readonly_props = [p._rest_name for p in self._attr_to_rest_field.values() if _is_readonly(p)]
1081 for k, v in self.items():
1082 if exclude_readonly and k in readonly_props: # pyright: ignore
1083 continue
1084 is_multipart_file_input = False
1085 try:
1086 is_multipart_file_input = next(
1087 rf for rf in self._attr_to_rest_field.values() if rf._rest_name == k
1088 )._is_multipart_file_input
1089 except StopIteration:
1090 pass
1091 result[k] = v if is_multipart_file_input else Model._as_dict_value(v, exclude_readonly=exclude_readonly)
1092 return result
1094 @staticmethod
1095 def _as_dict_value(v: typing.Any, exclude_readonly: bool = False) -> typing.Any:
1096 if v is None or isinstance(v, _Null):
1097 return None
1098 if isinstance(v, (list, tuple, set)):
1099 return type(v)(Model._as_dict_value(x, exclude_readonly=exclude_readonly) for x in v)
1100 if isinstance(v, dict):
1101 return {dk: Model._as_dict_value(dv, exclude_readonly=exclude_readonly) for dk, dv in v.items()}
1102 return v.as_dict(exclude_readonly=exclude_readonly) if hasattr(v, "as_dict") else v
1105def _deserialize_model(model_deserializer: typing.Optional[typing.Callable], obj):
1106 if _is_model(obj):
1107 return obj
1108 return _deserialize(model_deserializer, obj)
1111def _deserialize_with_optional(if_obj_deserializer: typing.Optional[typing.Callable], obj):
1112 if obj is None:
1113 return obj
1114 return _deserialize_with_callable(if_obj_deserializer, obj)
1117def _deserialize_with_union(deserializers, obj):
1118 for deserializer in deserializers:
1119 try:
1120 return _deserialize(deserializer, obj)
1121 except DeserializationError:
1122 pass
1123 raise DeserializationError()
1126def _deserialize_dict(
1127 value_deserializer: typing.Optional[typing.Callable],
1128 module: typing.Optional[str],
1129 obj: dict[typing.Any, typing.Any],
1130):
1131 if obj is None:
1132 return obj
1133 if isinstance(obj, ET.Element):
1134 obj = {child.tag: child for child in obj}
1135 return {k: _deserialize(value_deserializer, v, module) for k, v in obj.items()}
1138def _deserialize_multiple_sequence(
1139 entry_deserializers: list[typing.Optional[typing.Callable]],
1140 module: typing.Optional[str],
1141 obj,
1142):
1143 if obj is None:
1144 return obj
1145 return type(obj)(_deserialize(deserializer, entry, module) for entry, deserializer in zip(obj, entry_deserializers))
1148def _is_array_encoded_deserializer(deserializer: functools.partial) -> bool:
1149 return (
1150 isinstance(deserializer, functools.partial)
1151 and isinstance(deserializer.args[0], functools.partial)
1152 and deserializer.args[0].func == _deserialize_array_encoded # pylint: disable=comparison-with-callable
1153 )
1156def _deserialize_sequence(
1157 deserializer: typing.Optional[typing.Callable],
1158 module: typing.Optional[str],
1159 obj,
1160):
1161 if obj is None:
1162 return obj
1163 if isinstance(obj, ET.Element):
1164 obj = list(obj)
1166 # encoded string may be deserialized to sequence
1167 if isinstance(obj, str) and isinstance(deserializer, functools.partial):
1168 # for list[str]
1169 if _is_array_encoded_deserializer(deserializer):
1170 return deserializer(obj)
1172 # for list[Union[...]]
1173 if isinstance(deserializer.args[0], list):
1174 for sub_deserializer in deserializer.args[0]:
1175 if _is_array_encoded_deserializer(sub_deserializer):
1176 return sub_deserializer(obj)
1178 return type(obj)(_deserialize(deserializer, entry, module) for entry in obj)
1181def _sorted_annotations(types: list[typing.Any]) -> list[typing.Any]:
1182 return sorted(
1183 types,
1184 key=lambda x: hasattr(x, "__name__") and x.__name__.lower() in ("str", "float", "int", "bool"),
1185 )
1188def _get_deserialize_callable_from_annotation( # pylint: disable=too-many-return-statements, too-many-statements, too-many-branches
1189 annotation: typing.Any,
1190 module: typing.Optional[str],
1191 rf: typing.Optional["_RestField"] = None,
1192) -> typing.Optional[typing.Callable[[typing.Any], typing.Any]]:
1193 if not annotation:
1194 return None
1196 # is it a type alias?
1197 if isinstance(annotation, str):
1198 if module is not None:
1199 annotation = _get_type_alias_type(module, annotation)
1201 # is it a forward ref / in quotes?
1202 if isinstance(annotation, (str, typing.ForwardRef)):
1203 try:
1204 model_name = annotation.__forward_arg__ # type: ignore
1205 except AttributeError:
1206 model_name = annotation
1207 if module is not None:
1208 annotation = _get_model(module, model_name) # type: ignore
1210 try:
1211 if module and _is_model(annotation):
1212 if rf:
1213 rf._is_model = True
1215 return functools.partial(_deserialize_model, annotation) # pyright: ignore
1216 except Exception:
1217 pass
1219 # is it a literal?
1220 try:
1221 if annotation.__origin__ is typing.Literal: # pyright: ignore
1222 return None
1223 except AttributeError:
1224 pass
1226 # is it optional?
1227 try:
1228 if any(a is _NONE_TYPE for a in annotation.__args__): # pyright: ignore
1229 if rf:
1230 rf._is_optional = True
1231 if len(annotation.__args__) <= 2: # pyright: ignore
1232 if_obj_deserializer = _get_deserialize_callable_from_annotation(
1233 next(a for a in annotation.__args__ if a is not _NONE_TYPE), module, rf # pyright: ignore
1234 )
1236 return functools.partial(_deserialize_with_optional, if_obj_deserializer)
1237 # the type is Optional[Union[...]], we need to remove the None type from the Union
1238 annotation_copy = copy.copy(annotation)
1239 annotation_copy.__args__ = [a for a in annotation_copy.__args__ if a is not _NONE_TYPE] # pyright: ignore
1240 return _get_deserialize_callable_from_annotation(annotation_copy, module, rf)
1241 except AttributeError:
1242 pass
1244 # is it union?
1245 if getattr(annotation, "__origin__", None) is typing.Union:
1246 # initial ordering is we make `string` the last deserialization option, because it is often them most generic
1247 deserializers = [
1248 _get_deserialize_callable_from_annotation(arg, module, rf)
1249 for arg in _sorted_annotations(annotation.__args__) # pyright: ignore
1250 ]
1252 return functools.partial(_deserialize_with_union, deserializers)
1254 try:
1255 annotation_name = (
1256 annotation.__name__ if hasattr(annotation, "__name__") else annotation._name # pyright: ignore
1257 )
1258 if annotation_name.lower() == "dict":
1259 value_deserializer = _get_deserialize_callable_from_annotation(
1260 annotation.__args__[1], module, rf # pyright: ignore
1261 )
1263 return functools.partial(
1264 _deserialize_dict,
1265 value_deserializer,
1266 module,
1267 )
1268 except (AttributeError, IndexError):
1269 pass
1270 try:
1271 annotation_name = (
1272 annotation.__name__ if hasattr(annotation, "__name__") else annotation._name # pyright: ignore
1273 )
1274 if annotation_name.lower() in ["list", "set", "tuple", "sequence"]:
1275 if len(annotation.__args__) > 1: # pyright: ignore
1276 entry_deserializers = [
1277 _get_deserialize_callable_from_annotation(dt, module, rf)
1278 for dt in annotation.__args__ # pyright: ignore
1279 ]
1280 return functools.partial(_deserialize_multiple_sequence, entry_deserializers, module)
1281 deserializer = _get_deserialize_callable_from_annotation(
1282 annotation.__args__[0], module, rf # pyright: ignore
1283 )
1285 return functools.partial(_deserialize_sequence, deserializer, module)
1286 except (TypeError, IndexError, AttributeError, SyntaxError):
1287 pass
1289 def _deserialize_default(
1290 deserializer,
1291 obj,
1292 ):
1293 if obj is None:
1294 return obj
1295 try:
1296 return _deserialize_with_callable(deserializer, obj)
1297 except Exception:
1298 pass
1299 return obj
1301 if get_deserializer(annotation, rf):
1302 return functools.partial(_deserialize_default, get_deserializer(annotation, rf))
1304 return functools.partial(_deserialize_default, annotation)
1307def _deserialize_with_callable(
1308 deserializer: typing.Optional[typing.Callable[[typing.Any], typing.Any]],
1309 value: typing.Any,
1310): # pylint: disable=too-many-return-statements
1311 try:
1312 if value is None or isinstance(value, _Null):
1313 return None
1314 if isinstance(value, ET.Element):
1315 if deserializer is str:
1316 return value.text or ""
1317 if deserializer is int:
1318 return int(value.text) if value.text else None
1319 if deserializer is float:
1320 return float(value.text) if value.text else None
1321 if deserializer is bool:
1322 return value.text == "true" if value.text else None
1323 if deserializer and deserializer in _DESERIALIZE_MAPPING.values():
1324 return deserializer(value.text) if value.text else None
1325 if deserializer and deserializer in _DESERIALIZE_MAPPING_WITHFORMAT.values():
1326 return deserializer(value.text) if value.text else None
1327 if deserializer is None:
1328 return value
1329 if deserializer in [int, float, bool]:
1330 return deserializer(value)
1331 if isinstance(deserializer, CaseInsensitiveEnumMeta):
1332 try:
1333 return deserializer(value.text if isinstance(value, ET.Element) else value)
1334 except ValueError:
1335 # for unknown value, return raw value
1336 return value.text if isinstance(value, ET.Element) else value
1337 if isinstance(deserializer, type) and issubclass(deserializer, Model):
1338 return deserializer._deserialize(value, [])
1339 return typing.cast(typing.Callable[[typing.Any], typing.Any], deserializer)(value)
1340 except Exception as e:
1341 raise DeserializationError() from e
1344def _deserialize(
1345 deserializer: typing.Any,
1346 value: typing.Any,
1347 module: typing.Optional[str] = None,
1348 rf: typing.Optional["_RestField"] = None,
1349 format: typing.Optional[str] = None,
1350) -> typing.Any:
1351 if isinstance(value, PipelineResponse):
1352 value = value.http_response.json()
1353 if rf is None and format:
1354 rf = _RestField(format=format)
1355 if not isinstance(deserializer, functools.partial):
1356 deserializer = _get_deserialize_callable_from_annotation(deserializer, module, rf)
1357 return _deserialize_with_callable(deserializer, value)
1360def _failsafe_deserialize(
1361 deserializer: typing.Any,
1362 response: HttpResponse,
1363 module: typing.Optional[str] = None,
1364 rf: typing.Optional["_RestField"] = None,
1365 format: typing.Optional[str] = None,
1366) -> typing.Any:
1367 try:
1368 return _deserialize(deserializer, response.json(), module, rf, format)
1369 except Exception: # pylint: disable=broad-except
1370 _LOGGER.warning(
1371 "Ran into a deserialization error. Ignoring since this is failsafe deserialization", exc_info=True
1372 )
1373 return None
1376def _failsafe_deserialize_xml(
1377 deserializer: typing.Any,
1378 response: HttpResponse,
1379) -> typing.Any:
1380 try:
1381 return _deserialize_xml(deserializer, response.text())
1382 except Exception: # pylint: disable=broad-except
1383 _LOGGER.warning(
1384 "Ran into a deserialization error. Ignoring since this is failsafe deserialization", exc_info=True
1385 )
1386 return None
1389# pylint: disable=too-many-instance-attributes
1390class _RestField:
1391 def __init__(
1392 self,
1393 *,
1394 name: typing.Optional[str] = None,
1395 type: typing.Optional[typing.Callable] = None, # pylint: disable=redefined-builtin
1396 is_discriminator: bool = False,
1397 visibility: typing.Optional[list[str]] = None,
1398 default: typing.Any = _UNSET,
1399 format: typing.Optional[str] = None,
1400 is_multipart_file_input: bool = False,
1401 xml: typing.Optional[dict[str, typing.Any]] = None,
1402 deserializer: typing.Optional[typing.Callable] = None,
1403 ):
1404 self._type = type
1405 self._rest_name_input = name
1406 self._module: typing.Optional[str] = None
1407 self._is_discriminator = is_discriminator
1408 self._visibility = visibility
1409 self._is_model = False
1410 self._is_optional = False
1411 self._default = default
1412 self._format = format
1413 self._is_multipart_file_input = is_multipart_file_input
1414 self._xml = xml if xml is not None else {}
1415 self._deserializer = deserializer
1417 @property
1418 def _class_type(self) -> typing.Any:
1419 result = getattr(self._type, "args", [None])[0]
1420 # type may be wrapped by nested functools.partial so we need to check for that
1421 if isinstance(result, functools.partial):
1422 return getattr(result, "args", [None])[0]
1423 return result
1425 @property
1426 def _rest_name(self) -> str:
1427 if self._rest_name_input is None:
1428 raise ValueError("Rest name was never set")
1429 return self._rest_name_input
1431 def __get__(self, obj: Model, type=None): # pylint: disable=redefined-builtin
1432 # by this point, type and rest_name will have a value bc we default
1433 # them in __new__ of the Model class
1434 # Use _data.get() directly to avoid triggering __getitem__ which clears the cache
1435 item = obj._data.get(self._rest_name, _UNSET)
1436 if item is _UNSET:
1437 # Field not set by user; return the client default if one exists, otherwise None
1438 return self._default if self._default is not _UNSET else None
1439 if item is None:
1440 return item
1441 if self._is_model:
1442 return item
1444 # For mutable types, we want mutations to directly affect _data
1445 # Check if we've already deserialized this value
1446 cache_attr = f"_deserialized_{self._rest_name}"
1447 if hasattr(obj, cache_attr):
1448 # Return the value from _data directly (it's been deserialized in place)
1449 return obj._data.get(self._rest_name)
1451 # Fast path: use _deserializer directly (avoids _serialize/_deserialize chain)
1452 if self._deserializer:
1453 deserialized = self._deserializer(item)
1454 else:
1455 deserialized = _deserialize(self._type, _serialize(item, self._format), rf=self)
1457 # For mutable types, store the deserialized value back in _data
1458 # so mutations directly affect _data
1459 if isinstance(deserialized, (dict, list, set)):
1460 obj._data[self._rest_name] = deserialized
1461 object.__setattr__(obj, cache_attr, True) # Mark as deserialized
1462 return deserialized
1464 return deserialized
1466 def __set__(self, obj: Model, value) -> None:
1467 # Clear the cached deserialized object when setting a new value
1468 cache_attr = f"_deserialized_{self._rest_name}"
1469 if hasattr(obj, cache_attr):
1470 object.__delattr__(obj, cache_attr)
1472 if value is None:
1473 # we want to wipe out entries if users set attr to None
1474 try:
1475 obj.__delitem__(self._rest_name)
1476 except KeyError:
1477 pass
1478 return
1479 if self._is_model:
1480 if not _is_model(value):
1481 value = _deserialize(self._type, value)
1482 obj.__setitem__(self._rest_name, value)
1483 return
1484 obj.__setitem__(self._rest_name, _serialize(value, self._format))
1486 def _get_deserialize_callable_from_annotation(
1487 self, annotation: typing.Any
1488 ) -> typing.Optional[typing.Callable[[typing.Any], typing.Any]]:
1489 return _get_deserialize_callable_from_annotation(annotation, self._module, self)
1492def rest_field(
1493 *,
1494 name: typing.Optional[str] = None,
1495 type: typing.Optional[typing.Callable] = None, # pylint: disable=redefined-builtin
1496 visibility: typing.Optional[list[str]] = None,
1497 default: typing.Any = _UNSET,
1498 format: typing.Optional[str] = None,
1499 is_multipart_file_input: bool = False,
1500 xml: typing.Optional[dict[str, typing.Any]] = None,
1501 deserializer: typing.Optional[typing.Callable] = None,
1502) -> typing.Any:
1503 return _RestField(
1504 name=name,
1505 type=type,
1506 visibility=visibility,
1507 default=default,
1508 format=format,
1509 is_multipart_file_input=is_multipart_file_input,
1510 xml=xml,
1511 deserializer=deserializer,
1512 )
1515def rest_discriminator(
1516 *,
1517 name: typing.Optional[str] = None,
1518 type: typing.Optional[typing.Callable] = None, # pylint: disable=redefined-builtin
1519 visibility: typing.Optional[list[str]] = None,
1520 xml: typing.Optional[dict[str, typing.Any]] = None,
1521) -> typing.Any:
1522 return _RestField(name=name, type=type, is_discriminator=True, visibility=visibility, xml=xml)
1525def serialize_xml(model: Model, exclude_readonly: bool = False) -> str:
1526 """Serialize a model to XML.
1528 :param Model model: The model to serialize.
1529 :param bool exclude_readonly: Whether to exclude readonly properties.
1530 :returns: The XML representation of the model.
1531 :rtype: str
1532 """
1533 return ET.tostring(_get_element(model, exclude_readonly), encoding="unicode") # type: ignore
1536def _get_xml_ns(meta: dict[str, typing.Any]) -> typing.Optional[str]:
1537 """Return the XML namespace from a metadata dict, checking both 'ns' (old-style) and 'namespace' (DPG) keys.
1539 :param dict meta: The metadata dictionary to extract namespace from.
1540 :returns: The namespace string if 'ns' or 'namespace' key is present, None otherwise.
1541 :rtype: str or None
1542 """
1543 ns = meta.get("ns")
1544 if ns is None:
1545 ns = meta.get("namespace")
1546 return ns
1549def _resolve_xml_ns(
1550 prop_meta: dict[str, typing.Any], model_meta: typing.Optional[dict[str, typing.Any]] = None
1551) -> typing.Optional[str]:
1552 """Resolve XML namespace for a property, falling back to model namespace when appropriate.
1554 Checks the property metadata first; if no namespace is found and the model does not declare
1555 an explicit prefix, falls back to the model-level namespace.
1557 :param dict prop_meta: The property metadata dictionary.
1558 :param dict model_meta: The model metadata dictionary, used as fallback.
1559 :returns: The resolved namespace string, or None.
1560 :rtype: str or None
1561 """
1562 ns = _get_xml_ns(prop_meta)
1563 if ns is None and model_meta is not None and not model_meta.get("prefix"):
1564 ns = _get_xml_ns(model_meta)
1565 return ns
1568def _set_xml_attribute(element: ET.Element, name: str, value: typing.Any, prop_meta: dict[str, typing.Any]) -> None:
1569 """Set an XML attribute on an element, handling namespace prefix registration.
1571 :param ET.Element element: The element to set the attribute on.
1572 :param str name: The default attribute name (wire name).
1573 :param any value: The attribute value.
1574 :param dict prop_meta: The property metadata dictionary.
1575 """
1576 xml_name = prop_meta.get("name", name)
1577 _attr_ns = _get_xml_ns(prop_meta)
1578 if _attr_ns:
1579 _attr_prefix = prop_meta.get("prefix")
1580 if _attr_prefix:
1581 _safe_register_namespace(_attr_prefix, _attr_ns)
1582 xml_name = "{" + _attr_ns + "}" + xml_name
1583 element.set(xml_name, _get_primitive_type_value(value))
1586def _get_element(
1587 o: typing.Any,
1588 exclude_readonly: bool = False,
1589 parent_meta: typing.Optional[dict[str, typing.Any]] = None,
1590 wrapped_element: typing.Optional[ET.Element] = None,
1591) -> typing.Union[ET.Element, list[ET.Element]]:
1592 if _is_model(o):
1593 model_meta = getattr(o, "_xml", {})
1595 # if prop is a model, then use the prop element directly, else generate a wrapper of model
1596 if wrapped_element is None:
1597 # When serializing as an array item (parent_meta is set), check if the parent has an
1598 # explicit itemsName. This ensures correct element names for unwrapped arrays (where
1599 # the element tag is the property/items name, not the model type name).
1600 _items_name = parent_meta.get("itemsName") if parent_meta is not None else None
1601 element_name = _items_name if _items_name else (model_meta.get("name") or o.__class__.__name__)
1602 _model_ns = _get_xml_ns(model_meta)
1603 wrapped_element = _create_xml_element(
1604 element_name,
1605 model_meta.get("prefix"),
1606 _model_ns,
1607 )
1609 readonly_props = []
1610 if exclude_readonly:
1611 readonly_props = [p._rest_name for p in o._attr_to_rest_field.values() if _is_readonly(p)]
1613 for k, v in o.items():
1614 # do not serialize readonly properties
1615 if exclude_readonly and k in readonly_props:
1616 continue
1618 prop_rest_field = _get_rest_field(o._attr_to_rest_field, k)
1619 if prop_rest_field:
1620 prop_meta = getattr(prop_rest_field, "_xml").copy()
1621 # use the wire name as xml name if no specific name is set
1622 if prop_meta.get("name") is None:
1623 prop_meta["name"] = k
1624 else:
1625 # additional properties will not have rest field, use the wire name as xml name
1626 prop_meta = {"name": k}
1628 # Propagate model namespace to properties only for old-style "ns"-keyed models.
1629 # DPG-generated models use the "namespace" key and explicitly declare namespace on
1630 # each property that needs it, so propagation is intentionally skipped for them.
1631 if prop_meta.get("ns") is None and model_meta.get("ns"):
1632 prop_meta["ns"] = model_meta.get("ns")
1633 prop_meta["prefix"] = model_meta.get("prefix")
1635 if prop_meta.get("unwrapped", False):
1636 # unwrapped could only set on array
1637 wrapped_element.extend(_get_element(v, exclude_readonly, prop_meta))
1638 elif prop_meta.get("text", False):
1639 # text could only set on primitive type
1640 wrapped_element.text = _get_primitive_type_value(v)
1641 elif prop_meta.get("attribute", False):
1642 _set_xml_attribute(wrapped_element, k, v, prop_meta)
1643 else:
1644 # other wrapped prop element
1645 wrapped_element.append(_get_wrapped_element(v, exclude_readonly, prop_meta))
1646 return wrapped_element
1647 if isinstance(o, list):
1648 return [_get_element(x, exclude_readonly, parent_meta) for x in o] # type: ignore
1649 if isinstance(o, dict):
1650 result = []
1651 _dict_ns = _get_xml_ns(parent_meta) if parent_meta else None
1652 for k, v in o.items():
1653 result.append(
1654 _get_wrapped_element(
1655 v,
1656 exclude_readonly,
1657 {
1658 "name": k,
1659 "ns": _dict_ns,
1660 "prefix": parent_meta.get("prefix") if parent_meta else None,
1661 },
1662 )
1663 )
1664 return result
1666 # primitive case need to create element based on parent_meta
1667 if parent_meta:
1668 _items_ns = parent_meta.get("itemsNs")
1669 if _items_ns is None:
1670 _items_ns = _get_xml_ns(parent_meta)
1671 return _get_wrapped_element(
1672 o,
1673 exclude_readonly,
1674 {
1675 "name": parent_meta.get("itemsName", parent_meta.get("name")),
1676 "prefix": parent_meta.get("itemsPrefix", parent_meta.get("prefix")),
1677 "ns": _items_ns,
1678 },
1679 )
1681 raise ValueError("Could not serialize value into xml: " + o)
1684def _get_wrapped_element(
1685 v: typing.Any,
1686 exclude_readonly: bool,
1687 meta: typing.Optional[dict[str, typing.Any]],
1688) -> ET.Element:
1689 _meta_ns = _get_xml_ns(meta) if meta else None
1690 wrapped_element = _create_xml_element(
1691 meta.get("name") if meta else None, meta.get("prefix") if meta else None, _meta_ns
1692 )
1693 if isinstance(v, (dict, list)):
1694 wrapped_element.extend(_get_element(v, exclude_readonly, meta))
1695 elif _is_model(v):
1696 _get_element(v, exclude_readonly, meta, wrapped_element)
1697 else:
1698 wrapped_element.text = _get_primitive_type_value(v)
1699 return wrapped_element # type: ignore[no-any-return]
1702def _get_primitive_type_value(v) -> str:
1703 if v is True:
1704 return "true"
1705 if v is False:
1706 return "false"
1707 if isinstance(v, _Null):
1708 return ""
1709 return str(v)
1712def _safe_register_namespace(prefix: str, ns: str) -> None:
1713 """Register an XML namespace prefix, handling reserved prefix patterns.
1715 Some prefixes (e.g. 'ns2') match Python's reserved 'ns\\d+' pattern used for
1716 auto-generated prefixes, causing register_namespace to raise ValueError.
1717 Falls back to directly registering in the internal namespace map.
1719 :param str prefix: The namespace prefix to register.
1720 :param str ns: The namespace URI.
1721 """
1722 try:
1723 ET.register_namespace(prefix, ns)
1724 except ValueError:
1725 _ns_map = getattr(ET, "_namespace_map", None)
1726 if _ns_map is not None:
1727 _ns_map[ns] = prefix
1730def _create_xml_element(
1731 tag: typing.Any, prefix: typing.Optional[str] = None, ns: typing.Optional[str] = None
1732) -> ET.Element:
1733 if prefix and ns:
1734 _safe_register_namespace(prefix, ns)
1735 if ns:
1736 return ET.Element("{" + ns + "}" + tag)
1737 return ET.Element(tag)
1740def _deserialize_xml(
1741 deserializer: typing.Any,
1742 value: str,
1743) -> typing.Any:
1744 element = ET.fromstring(value) # nosec
1745 if _is_model(deserializer):
1746 return deserializer._deserialize(element, [])
1747 return _deserialize(deserializer, element)
1750def _convert_element(e: ET.Element):
1751 # dict case
1752 if len(e.attrib) > 0 or len({child.tag for child in e}) > 1:
1753 dict_result: dict[str, typing.Any] = {}
1754 for child in e:
1755 if dict_result.get(child.tag) is not None:
1756 if isinstance(dict_result[child.tag], list):
1757 dict_result[child.tag].append(_convert_element(child))
1758 else:
1759 dict_result[child.tag] = [dict_result[child.tag], _convert_element(child)]
1760 else:
1761 dict_result[child.tag] = _convert_element(child)
1762 dict_result.update(e.attrib)
1763 return dict_result
1764 # array case
1765 if len(e) > 0:
1766 array_result: list[typing.Any] = []
1767 for child in e:
1768 array_result.append(_convert_element(child))
1769 return array_result
1770 # primitive case
1771 return e.text