CodeGen from PR 29328 in Azure/azure-rest-api-specs

Merge 03e89671b50de6a14eaa0a3c5c579d6cf057889a into f6fbf63641332dd06c025d995a82dc090a5f6fad

sdk/communication/azure-communication-jobrouter/_meta.json

@@ -0,0 +1,7 @@
+{
+  "commit": "60d66ff70e2c025e2ffe107cb79723dc59c44a80",
+  "repository_url": "https://github.com/Azure/azure-rest-api-specs",
+  "typespec_src": "specification/communication/Communication.JobRouter",
+  "@azure-tools/typespec-python": "0.23.13",
+  "@autorest/python": "6.13.18"
+}

sdk/communication/azure-communication-jobrouter/azure/communication/jobrouter/__init__.py

@@ -6,8 +6,8 @@
 # Changes may cause incorrect behavior and will be lost if the code is regenerated.
 # --------------------------------------------------------------------------
 
-from ._patch import JobRouterAdministrationClient
-from ._patch import JobRouterClient
+from ._client import JobRouterAdministrationClient
+from ._client import JobRouterClient
 from ._version import VERSION
 
 __version__ = VERSION

sdk/communication/azure-communication-jobrouter/azure/communication/jobrouter/_client.py

@@ -79,7 +79,7 @@ def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs:
 
         request_copy = deepcopy(request)
         path_format_arguments = {
-            "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str", skip_quote=True),
+            "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str"),
         }
 
         request_copy.url = self._client.format_url(request_copy.url, **path_format_arguments)
@@ -155,7 +155,7 @@ def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs:
 
         request_copy = deepcopy(request)
         path_format_arguments = {
-            "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str", skip_quote=True),
+            "endpoint": self._serialize.url("self._config.endpoint", self._config.endpoint, "str"),
         }
 
         request_copy.url = self._client.format_url(request_copy.url, **path_format_arguments)

sdk/communication/azure-communication-jobrouter/azure/communication/jobrouter/_model_base.py

@@ -6,14 +6,14 @@
 # --------------------------------------------------------------------------
 # pylint: disable=protected-access, arguments-differ, signature-differs, broad-except
 
+import copy
 import calendar
 import decimal
 import functools
 import sys
 import logging
 import base64
 import re
-import copy
 import typing
 import enum
 import email.utils
@@ -339,7 +339,7 @@ def _get_model(module_name: str, model_name: str):
 
 class _MyMutableMapping(MutableMapping[str, typing.Any]):  # pylint: disable=unsubscriptable-object
     def __init__(self, data: typing.Dict[str, typing.Any]) -> None:
-        self._data = copy.deepcopy(data)
+        self._data = data
 
     def __contains__(self, key: typing.Any) -> bool:
         return key in self._data
@@ -378,16 +378,13 @@ def get(self, key: str, default: typing.Any = None) -> typing.Any:
             return default
 
     @typing.overload
-    def pop(self, key: str) -> typing.Any:
-        ...
+    def pop(self, key: str) -> typing.Any: ...
 
     @typing.overload
-    def pop(self, key: str, default: _T) -> _T:
-        ...
+    def pop(self, key: str, default: _T) -> _T: ...
 
     @typing.overload
-    def pop(self, key: str, default: typing.Any) -> typing.Any:
-        ...
+    def pop(self, key: str, default: typing.Any) -> typing.Any: ...
 
     def pop(self, key: str, default: typing.Any = _UNSET) -> typing.Any:
         if default is _UNSET:
@@ -404,12 +401,10 @@ def update(self, *args: typing.Any, **kwargs: typing.Any) -> None:
         self._data.update(*args, **kwargs)
 
     @typing.overload
-    def setdefault(self, key: str, default: None = None) -> None:
-        ...
+    def setdefault(self, key: str, default: None = None) -> None: ...
 
     @typing.overload
-    def setdefault(self, key: str, default: typing.Any) -> typing.Any:
-        ...
+    def setdefault(self, key: str, default: typing.Any) -> typing.Any: ...
 
     def setdefault(self, key: str, default: typing.Any = _UNSET) -> typing.Any:
         if default is _UNSET:
@@ -594,6 +589,64 @@ def _as_dict_value(v: typing.Any, exclude_readonly: bool = False) -> typing.Any:
         return v.as_dict(exclude_readonly=exclude_readonly) if hasattr(v, "as_dict") else v
 
 
+def _deserialize_model(model_deserializer: typing.Optional[typing.Callable], obj):
+    if _is_model(obj):
+        return obj
+    return _deserialize(model_deserializer, obj)
+
+
+def _deserialize_with_optional(if_obj_deserializer: typing.Optional[typing.Callable], obj):
+    if obj is None:
+        return obj
+    return _deserialize_with_callable(if_obj_deserializer, obj)
+
+
+def _deserialize_with_union(deserializers, obj):
+    for deserializer in deserializers:
+        try:
+            return _deserialize(deserializer, obj)
+        except DeserializationError:
+            pass
+    raise DeserializationError()
+
+
+def _deserialize_dict(
+    value_deserializer: typing.Optional[typing.Callable],
+    module: typing.Optional[str],
+    obj: typing.Dict[typing.Any, typing.Any],
+):
+    if obj is None:
+        return obj
+    return {k: _deserialize(value_deserializer, v, module) for k, v in obj.items()}
+
+
+def _deserialize_multiple_sequence(
+    entry_deserializers: typing.List[typing.Optional[typing.Callable]],
+    module: typing.Optional[str],
+    obj,
+):
+    if obj is None:
+        return obj
+    return type(obj)(_deserialize(deserializer, entry, module) for entry, deserializer in zip(obj, entry_deserializers))
+
+
+def _deserialize_sequence(
+    deserializer: typing.Optional[typing.Callable],
+    module: typing.Optional[str],
+    obj,
+):
+    if obj is None:
+        return obj
+    return type(obj)(_deserialize(deserializer, entry, module) for entry in obj)
+
+
+def _sorted_annotations(types: typing.List[typing.Any]) -> typing.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=R0911, R0915, R0912
     annotation: typing.Any,
     module: typing.Optional[str],
@@ -621,11 +674,6 @@ def _get_deserialize_callable_from_annotation(  # pylint: disable=R0911, R0915,
             if rf:
                 rf._is_model = True
 
-            def _deserialize_model(model_deserializer: typing.Optional[typing.Callable], obj):
-                if _is_model(obj):
-                    return obj
-                return _deserialize(model_deserializer, obj)
-
             return functools.partial(_deserialize_model, annotation)  # pyright: ignore
     except Exception:
         pass
@@ -640,36 +688,27 @@ def _deserialize_model(model_deserializer: typing.Optional[typing.Callable], obj
     # is it optional?
     try:
         if any(a for a in annotation.__args__ if a == type(None)):  # pyright: ignore
-            if_obj_deserializer = _get_deserialize_callable_from_annotation(
-                next(a for a in annotation.__args__ if a != type(None)), module, rf  # pyright: ignore
-            )
-
-            def _deserialize_with_optional(if_obj_deserializer: typing.Optional[typing.Callable], obj):
-                if obj is None:
-                    return obj
-                return _deserialize_with_callable(if_obj_deserializer, obj)
-
-            return functools.partial(_deserialize_with_optional, if_obj_deserializer)
+            if len(annotation.__args__) <= 2:  # pyright: ignore
+                if_obj_deserializer = _get_deserialize_callable_from_annotation(
+                    next(a for a in annotation.__args__ if a != type(None)), module, rf  # pyright: ignore
+                )
+
+                return functools.partial(_deserialize_with_optional, if_obj_deserializer)
+            # the type is Optional[Union[...]], we need to remove the None type from the Union
+            annotation_copy = copy.copy(annotation)
+            annotation_copy.__args__ = [a for a in annotation_copy.__args__ if a != type(None)]  # pyright: ignore
+            return _get_deserialize_callable_from_annotation(annotation_copy, module, rf)
     except AttributeError:
         pass
 
+    # is it union?
     if getattr(annotation, "__origin__", None) is typing.Union:
         # initial ordering is we make `string` the last deserialization option, because it is often them most generic
         deserializers = [
             _get_deserialize_callable_from_annotation(arg, module, rf)
-            for arg in sorted(
-                annotation.__args__, key=lambda x: hasattr(x, "__name__") and x.__name__ == "str"  # pyright: ignore
-            )
+            for arg in _sorted_annotations(annotation.__args__)  # pyright: ignore
         ]
 
-        def _deserialize_with_union(deserializers, obj):
-            for deserializer in deserializers:
-                try:
-                    return _deserialize(deserializer, obj)
-                except DeserializationError:
-                    pass
-            raise DeserializationError()
-
         return functools.partial(_deserialize_with_union, deserializers)
 
     try:
@@ -678,53 +717,27 @@ def _deserialize_with_union(deserializers, obj):
                 annotation.__args__[1], module, rf  # pyright: ignore
             )
 
-            def _deserialize_dict(
-                value_deserializer: typing.Optional[typing.Callable],
-                obj: typing.Dict[typing.Any, typing.Any],
-            ):
-                if obj is None:
-                    return obj
-                return {k: _deserialize(value_deserializer, v, module) for k, v in obj.items()}
-
             return functools.partial(
                 _deserialize_dict,
                 value_deserializer,
+                module,
             )
     except (AttributeError, IndexError):
         pass
     try:
         if annotation._name in ["List", "Set", "Tuple", "Sequence"]:  # pyright: ignore
             if len(annotation.__args__) > 1:  # pyright: ignore
 
-                def _deserialize_multiple_sequence(
-                    entry_deserializers: typing.List[typing.Optional[typing.Callable]],
-                    obj,
-                ):
-                    if obj is None:
-                        return obj
-                    return type(obj)(
-                        _deserialize(deserializer, entry, module)
-                        for entry, deserializer in zip(obj, entry_deserializers)
-                    )
-
                 entry_deserializers = [
                     _get_deserialize_callable_from_annotation(dt, module, rf)
                     for dt in annotation.__args__  # pyright: ignore
                 ]
-                return functools.partial(_deserialize_multiple_sequence, entry_deserializers)
+                return functools.partial(_deserialize_multiple_sequence, entry_deserializers, module)
             deserializer = _get_deserialize_callable_from_annotation(
                 annotation.__args__[0], module, rf  # pyright: ignore
             )
 
-            def _deserialize_sequence(
-                deserializer: typing.Optional[typing.Callable],
-                obj,
-            ):
-                if obj is None:
-                    return obj
-                return type(obj)(_deserialize(deserializer, entry, module) for entry in obj)
-
-            return functools.partial(_deserialize_sequence, deserializer)
+            return functools.partial(_deserialize_sequence, deserializer, module)
     except (TypeError, IndexError, AttributeError, SyntaxError):
         pass
 

sdk/communication/azure-communication-jobrouter/azure/communication/jobrouter/_operations/__init__.py

@@ -6,8 +6,8 @@
 # Changes may cause incorrect behavior and will be lost if the code is regenerated.
 # --------------------------------------------------------------------------
 
-from ._patch import JobRouterAdministrationClientOperationsMixin
-from ._patch import JobRouterClientOperationsMixin
+from ._operations import JobRouterAdministrationClientOperationsMixin
+from ._operations import JobRouterClientOperationsMixin
 
 from ._patch import __all__ as _patch_all
 from ._patch import *  # pylint: disable=unused-wildcard-import