demo (DO NOT MREGE)

packages/typespec-python/test/azure/generated/authentication-api-key/authentication/apikey/_utils/model_base.py

@@ -162,6 +162,8 @@ def default(self, o):  # pylint: disable=too-many-return-statements
             except AttributeError:
                 # This will be raised when it hits value.total_seconds in the method above
                 pass
+            if _SERIALIZE_REGISTER.get(type(o)):
+                return _SERIALIZE_REGISTER.get(type(o))(o)
             return super(SdkJSONEncoder, self).default(o)
 
 
@@ -312,12 +314,17 @@ def _deserialize_int_as_str(attr):
 }
 
 
+_SERIALIZE_REGISTER = {}
+_DESERIALIZE_REGISTER = {}
+
 def get_deserializer(annotation: typing.Any, rf: typing.Optional["_RestField"] = None):
     if annotation is int and rf and rf._format == "str":
         return _deserialize_int_as_str
     if rf and rf._format:
         return _DESERIALIZE_MAPPING_WITHFORMAT.get(rf._format)
-    return _DESERIALIZE_MAPPING.get(annotation)  # pyright: ignore
+    if _DESERIALIZE_MAPPING.get(annotation):
+        return _DESERIALIZE_MAPPING.get(annotation)
+    return _DESERIALIZE_REGISTER.get(annotation)
 
 
 def _get_type_alias_type(module_name: str, alias_name: str):
@@ -511,6 +518,8 @@ def _serialize(o, format: typing.Optional[str] = None):  # pylint: disable=too-m
     except AttributeError:
         # This will be raised when it hits value.total_seconds in the method above
         pass
+    if _SERIALIZE_REGISTER.get(type(o)):
+        return _SERIALIZE_REGISTER.get(type(o))(o)
     return o
 
 

packages/typespec-python/test/azure/generated/authentication-api-key/authentication/apikey/models/_models.py

@@ -7,6 +7,7 @@
 # --------------------------------------------------------------------------
 # pylint: disable=useless-super-delegation
 
+from azure.core.messaging import CloudEvent
 from typing import Any, Mapping, overload
 
 from .._utils.model_base import Model as _Model, rest_field
@@ -20,13 +21,15 @@ class InvalidAuth(_Model):
     """
 
     error: str = rest_field(visibility=["read", "create", "update", "delete", "query"])
+    cloud_event: CloudEvent = rest_field(name="cloudEvent", visibility=["read", "create", "update", "delete", "query"])
     """Required."""
 
     @overload
     def __init__(
         self,
         *,
         error: str,
+        cloud_event: CloudEvent,
     ) -> None: ...
 
     @overload

packages/typespec-python/test/azure/generated/authentication-api-key/authentication/apikey/models/_patch.py

@@ -8,14 +8,29 @@
 Follow our quickstart for examples: https://aka.ms/azsdk/python/dpcodegen/python/customize
 """
 from typing import List
+from azure.core.messaging import CloudEvent
+
+from .._utils.model_base import _DESERIALIZE_REGISTER, _SERIALIZE_REGISTER
 
 __all__: List[str] = []  # Add all objects you want publicly available to users at this package level
 
 
+def cloud_event_serialize(obj) -> dict:
+    return {
+        "source": obj.source,
+        "type": obj.type,
+    }
+
+def cloud_event_deserialize(data: dict) -> CloudEvent:
+    return CloudEvent(source=data["source"], type=data["type"])
+
 def patch_sdk():
     """Do not remove from this file.
 
     `patch_sdk` is a last resort escape hatch that allows you to do customizations
     you can't accomplish using the techniques described in
     https://aka.ms/azsdk/python/dpcodegen/python/customize
     """
+
+    _SERIALIZE_REGISTER[CloudEvent] = cloud_event_serialize
+    _DESERIALIZE_REGISTER[CloudEvent] = cloud_event_deserialize

packages/typespec-python/test/azure/generated/authentication-api-key/test.py

@@ -0,0 +1,25 @@
+import json
+from authentication.apikey.models import InvalidAuth
+from authentication.apikey._utils.model_base import SdkJSONEncoder, _deserialize
+from azure.core.messaging import CloudEvent
+
+cloud_event = CloudEvent(source="source", type="type")
+
+# serialize
+print("========= serialize =================")
+model = InvalidAuth(error="error", cloud_event=cloud_event)
+print(model.cloud_event.source)
+print(model.cloud_event.type)
+print(json.dumps(model, cls=SdkJSONEncoder))
+
+# deserialize (external type as model property) 
+print("========= deserialize (external type as model property) =================")
+model_from_dict = _deserialize(InvalidAuth, {"error": "error", "cloudEvent": {"source": "source", "type": "type"}})
+print(model_from_dict.cloud_event.source)
+print(model_from_dict.cloud_event.type)
+
+# deserialize external type directly
+print("========= deserialize external type directly =================")
+cloud_event_from_dict = _deserialize(CloudEvent, {"source": "source", "type": "type"})
+print(cloud_event_from_dict.source)
+print(cloud_event_from_dict.type)