[Python] - Improve Model Deserialization Perf

packages/http-client-python/generator/pygen/codegen/templates/model_base.py.jinja2

@@ -662,6 +662,61 @@ def _create_value(rf: typing.Optional["_RestField"], value: typing.Any) -> typin
     return _serialize(value, rf._format)
 
 
+def _clone_wire_value(value: typing.Any) -> typing.Any:
+    if isinstance(value, list):
+        return [_clone_wire_value(entry) for entry in value]
+    if isinstance(value, dict):
+        return {key: _clone_wire_value(entry) for key, entry in value.items()}
+    if isinstance(value, set):
+        return {_clone_wire_value(entry) for entry in value}
+    if isinstance(value, tuple):
+        return tuple(_clone_wire_value(entry) for entry in value)
+    return value
+
+
+def _create_public_value(rf: typing.Optional["_RestField"], value: typing.Any) -> typing.Any:
+    if rf and rf._is_model:
+        value = _clone_wire_value(value)
+    return _create_value(rf, value)
+
+
+class _OwnedWireValue:
+    __slots__ = ("value",)
+
+    def __init__(self, value: typing.Mapping[str, typing.Any]) -> None:
+        self.value = value
+
+
+def _construct_from_wire(cls: type, data: typing.Any) -> typing.Any:
+    # ET.Element payloads are passed through as-is; JSON payloads are wrapped in
+    # _OwnedWireValue so the constructed model takes zero-copy ownership of the wire data.
+    return cls(data) if isinstance(data, ET.Element) else cls(_OwnedWireValue(data))
+
+
+def _create_value_from_wire(rf: typing.Optional["_RestField"], value: typing.Any) -> typing.Any:
+    """Build a stored value from an already-serialized (wire/JSON) payload.
+
+    Non-model values are already in wire form and are stored as-is; model fields and
+    ``ET.Element`` values are deserialized into their target type.
+
+    :param rf: The rest field describing the target attribute, if known.
+    :type rf: ~_RestField or None
+    :param value: The already-serialized value from the response body.
+    :type value: any
+    :return: The value to store in the model's backing dict.
+    :rtype: any
+    """
+    if not rf:
+        return _serialize(value, None)
+    if rf._is_multipart_file_input:
+        return value
+    if rf._is_model:
+        return _deserialize(rf._type, value)
+    if isinstance(value, ET.Element):
+        return _deserialize(rf._type, value)
+    return value
+
+
 # ============================================================================
 # Fast-path scalar deserializer functions for rest_field(deserializer=...)
 # These are referenced from rest_field declarations to bypass the generic
@@ -910,8 +965,18 @@ class Model(_MyMutableMapping):
             if isinstance(args[0], ET.Element):
                 dict_to_pass.update(self._init_from_xml(args[0]))
             else:
+                rest_field_by_rest_name = self._rest_field_by_rest_name
+                mapping = args[0]
+                if isinstance(mapping, _OwnedWireValue):
+                    mapping = mapping.value
+                    create_value = _create_value_from_wire
+                else:
+                    create_value = _create_public_value
                 dict_to_pass.update(
-                    {k: _create_value(_get_rest_field(self._attr_to_rest_field, k), v) for k, v in args[0].items()}
+                    {
+                        k: create_value(rest_field_by_rest_name.get(k), v)
+                        for k, v in mapping.items()
+                    }
                 )
         else:
             non_attr_kwargs = [k for k in kwargs if k not in self._attr_to_rest_field]
@@ -927,9 +992,7 @@ class Model(_MyMutableMapping):
             )
         # Apply client default values for fields the caller didn't set so that
         # defaults are part of `_data` and therefore included during serialization.
-        for rf in self._attr_to_rest_field.values():
-            if rf._default is _UNSET:
-                continue
+        for rf in self._fields_with_defaults:
             if rf._rest_name in dict_to_pass:
                 continue
             dict_to_pass[rf._rest_name] = _create_value(rf, rf._default)
@@ -1060,6 +1123,14 @@ class Model(_MyMutableMapping):
                 if not rf._rest_name_input:
                     rf._rest_name_input = attr
             cls._attr_to_rest_field: dict[str, _RestField] = dict(attr_to_rest_field.items())
+            # Mapping of rest_name -> _RestField, built once per class.
+            cls._rest_field_by_rest_name: dict[str, _RestField] = {
+                rf._rest_name: rf for rf in attr_to_rest_field.values()
+            }
+            # Subset of fields that declare a client-side default, built once per class.
+            cls._fields_with_defaults: list[_RestField] = [
+                rf for rf in attr_to_rest_field.values() if rf._default is not _UNSET
+            ]
             {% if code_model.has_padded_model_property %}
             cls._backcompat_attr_to_rest_field: dict[str, _RestField] = {
                 Model._get_backcompat_attribute_name(cls._attr_to_rest_field, attr): rf for attr, rf in cls
@@ -1100,10 +1171,10 @@ class Model(_MyMutableMapping):
     @classmethod
     def _deserialize(cls, data, exist_discriminators):
         if not hasattr(cls, "__mapping__"):
-            return cls(data)
+            return _construct_from_wire(cls, data)
         discriminator = cls._get_discriminator(exist_discriminators)
         if discriminator is None:
-            return cls(data)
+            return _construct_from_wire(cls, data)
         exist_discriminators.append(discriminator._rest_name)
         if isinstance(data, ET.Element):
             model_meta = getattr(cls, "_xml", {})
@@ -1234,13 +1305,82 @@ def _deserialize_sequence(
     return type(obj)(_deserialize(deserializer, entry, module) for entry in obj)
 
 
+_PRIMITIVE_SEQUENCE_TYPES = (int, float)
+
+
+def _deserialize_primitive_sequence(
+    builtin: typing.Callable,
+    deserializer: typing.Optional[typing.Callable],
+    module: typing.Optional[str],
+    obj,
+):
+    """Deserialize a homogeneous sequence of scalars.
+
+    Plain ``list``/``tuple``/``set`` inputs are converted with ``map(builtin, obj)``, falling
+    back to a per-element conversion that returns the raw entry when it cannot be converted.
+    Any other shape (encoded ``str``, ``ET.Element``, ...) is delegated to
+    :func:`_deserialize_sequence`.
+
+    :param builtin: The scalar constructor to apply (``int`` / ``float``).
+    :type builtin: callable
+    :param deserializer: The generic element deserializer used for the delegated path.
+    :type deserializer: callable or None
+    :param module: The module name used for forward-ref resolution in the delegated path.
+    :type module: str or None
+    :param obj: The already-parsed value.
+    :type obj: any
+    :return: The converted sequence.
+    :rtype: any
+    """
+    if obj is None:
+        return obj
+    if isinstance(obj, (list, tuple, set)):
+        try:
+            return type(obj)(map(builtin, obj))
+        except (TypeError, ValueError):
+
+            def _lenient(entry: typing.Any) -> typing.Any:
+                if entry is None:
+                    return entry
+                try:
+                    return builtin(entry)
+                except (TypeError, ValueError):
+                    return entry
+
+            return type(obj)(_lenient(entry) for entry in obj)
+    return _deserialize_sequence(deserializer, module, obj)
+
+
 def _sorted_annotations(types: list[typing.Any]) -> list[typing.Any]:
     return sorted(
         types,
         key=lambda x: hasattr(x, "__name__") and x.__name__.lower() in ("str", "float", "int", "bool"),
     )
 
-def _get_deserialize_callable_from_annotation(  # pylint: disable=too-many-return-statements, too-many-statements, too-many-branches
+@functools.lru_cache(maxsize=None)
+def _deserialize_callable_from_annotation_cached(
+    annotation: typing.Any,
+    module: typing.Optional[str],
+) -> typing.Optional[typing.Callable[[typing.Any], typing.Any]]:
+    return _resolve_deserialize_callable_from_annotation(annotation, module, None)
+
+
+def _get_deserialize_callable_from_annotation(
+    annotation: typing.Any,
+    module: typing.Optional[str],
+    rf: typing.Optional["_RestField"] = None,
+) -> typing.Optional[typing.Callable[[typing.Any], typing.Any]]:
+    # The rf-bound path may mutate rf (e.g. rf._is_model) and is resolved every call.
+    # The rf-less path is side-effect free, so it is memoized on (annotation, module).
+    if rf is not None:
+        return _resolve_deserialize_callable_from_annotation(annotation, module, rf)
+    try:
+        return _deserialize_callable_from_annotation_cached(annotation, module)
+    except TypeError:  # annotation can't be hashed, so it can't be used as a cache key
+        return _resolve_deserialize_callable_from_annotation(annotation, module, None)
+
+
+def _resolve_deserialize_callable_from_annotation(  # pylint: disable=too-many-return-statements, too-many-statements, too-many-branches
     annotation: typing.Any,
     module: typing.Optional[str],
     rf: typing.Optional["_RestField"] = None,
@@ -1336,6 +1476,11 @@ def _get_deserialize_callable_from_annotation(  # pylint: disable=too-many-retur
             deserializer = _get_deserialize_callable_from_annotation(
                 annotation.__args__[0], module, rf  # pyright: ignore
             )
+            element_annotation = annotation.__args__[0]  # pyright: ignore
+            if element_annotation in _PRIMITIVE_SEQUENCE_TYPES and not (rf and rf._format):
+                return functools.partial(
+                    _deserialize_primitive_sequence, element_annotation, deserializer, module
+                )
 
             return functools.partial(_deserialize_sequence, deserializer, module)
     except (TypeError, IndexError, AttributeError, SyntaxError):