GSOC : Add JSON serialization support for Python Dubbo

src/dubbo/client.py

@@ -13,6 +13,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
 import threading
 from typing import Optional
 
@@ -30,6 +31,7 @@
     RpcTypes,
     SerializingFunction,
 )
+from dubbo.codec import DubboCodec
 from dubbo.url import URL
 
 __all__ = ["Client"]
@@ -64,9 +66,10 @@ def _initialize(self):
             protocol = extensionLoader.get_extension(Protocol, self._reference.protocol)()
 
             registry_config = self._dubbo.registry_config
+            print("config",registry_config)
 
             self._protocol = RegistryProtocol(registry_config, protocol) if self._dubbo.registry_config else protocol
-
+            print(self._protocol)
             # build url
             reference_url = self._reference.to_url()
             if registry_config:
@@ -81,13 +84,30 @@ def _initialize(self):
             self._invoker = self._protocol.refer(self._url)
 
             self._initialized = True
+
+    @staticmethod
+    def get_codec(**kwargs) -> tuple:
+        """
+        Get the method encode and decode
+        :return: tuple of the encode and decode method 
+        """
+        return DubboCodec.get_serializer_deserializer(**kwargs)
+
 
     def unary(
         self,
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         return self._callable(
             MethodDescriptor(
                 method_name=method_name,
@@ -102,7 +122,15 @@ def client_stream(
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         return self._callable(
             MethodDescriptor(
                 method_name=method_name,
@@ -117,7 +145,15 @@ def server_stream(
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         return self._callable(
             MethodDescriptor(
                 method_name=method_name,
@@ -132,7 +168,15 @@ def bi_stream(
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         # create method descriptor
         return self._callable(
             MethodDescriptor(
@@ -160,4 +204,4 @@ def _callable(self, method_descriptor: MethodDescriptor) -> RpcCallable:
         url.attributes[common_constants.METHOD_DESCRIPTOR_KEY] = method_descriptor
 
         # create proxy
-        return self._callable_factory.get_callable(self._invoker, url)
+        return self._callable_factory.get_callable(self._invoker, url)

src/dubbo/codec/__init__.py

@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from .dubbo_codec import DubboCodec
+
+__all__ = ["DubboCodec"]

src/dubbo/codec/base_codec.py

@@ -0,0 +1,30 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from abc import ABC, abstractmethod
+from typing import Any, Union
+
+class Codec(ABC):
+    """Base codec interface for encoding/decoding data"""
+
+    @abstractmethod
+    def encode(self, data: Any) -> bytes:
+        """Encode Python object to bytes"""
+        pass
+
+    @abstractmethod
+    def decode(self, data: bytes) -> Any:
+        """Decode bytes to Python object"""
+        pass

src/dubbo/codec/codec_registry.py

@@ -0,0 +1,56 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from .base_codec import Codec
+from dubbo.codec.json_codec import JsonCodec
+from typing import Type, Any, Callable
+from pydantic import BaseModel
+
+class CodecRegistry:
+    """Registry for managing different codec types"""
+
+    _codecs = {
+        'json': JsonCodec
+    }
+
+    @classmethod
+    def get_codec(cls, codec_type: str, **kwargs) -> Codec:
+        """Return codec instance"""
+        if codec_type not in cls._codecs:
+            raise ValueError(f"Unsupported codec type: {codec_type}. Available: {list(cls._codecs.keys())}")
+
+        codec_class = cls._codecs[codec_type]
+        return codec_class(**kwargs)
+
+    @classmethod
+    def get_encoder_decoder(
+        cls,
+        codec_type: str,
+        model_type: Type[BaseModel],
+        **kwargs
+    ) -> tuple[Callable[[Any], bytes], Callable[[bytes], Any]]:
+        """Return encoder and decoder functions"""
+        codec_instance = cls.get_codec(codec_type, model_type=model_type, **kwargs)
+        return codec_instance.encode, codec_instance.decode
+
+    @classmethod
+    def register_codec(cls, name: str, codec_class: type):
+        """Register a new codec type"""
+        cls._codecs[name] = codec_class
+
+    @classmethod
+    def list_available_codecs(cls) -> list[str]:
+        """List all available codec types"""
+        return list(cls._codecs.keys())

src/dubbo/codec/dubbo_codec.py

@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from abc import ABC, abstractmethod
+from typing import Any, Type, Optional, Callable
+from .base_codec import Codec
+from .codec_registry import CodecRegistry
+from pydantic import BaseModel
+
+class DubboCodec:
+    _codec_instance: Optional[Codec] = None
+
+    @staticmethod
+    def init(codec_type: str = 'json', model_type: Optional[Type[BaseModel]] = None, **codec_kwargs):
+        """Initialize codec with specified type and options"""
+        if model_type is None:
+            raise ValueError("model_type is required for all codecs")
+
+        DubboCodec._codec_instance = CodecRegistry.get_codec(
+            codec_type, 
+            model_type=model_type, 
+            **codec_kwargs
+        )
+
+    @staticmethod
+    def get_instance() -> Codec:
+        if DubboCodec._codec_instance is None:
+            raise RuntimeError("DubboCodec is not initialized. Call DubboCodec.init(...) first.")
+        return DubboCodec._codec_instance
+
+    @staticmethod
+    def encode(data: Any) -> bytes:
+        return DubboCodec.get_instance().encode(data)
+
+    @staticmethod
+    def decode(data: bytes) -> Any:
+        return DubboCodec.get_instance().decode(data)
+
+    @staticmethod
+    def get_serializer_deserializer(
+        codec_type: str, 
+        request_model: Type[BaseModel] = None, 
+        response_model: Type[BaseModel] = None,
+        **codec_kwargs
+    ) -> tuple[Callable, Callable]:
+        """Get serializer and deserializer functions for RPC"""
+
+        request_codec = CodecRegistry.get_codec(codec_type, model_type=request_model, **codec_kwargs)
+
+        response_codec = CodecRegistry.get_codec(codec_type, model_type=response_model, **codec_kwargs)
+
+        def request_deserializer(data: bytes):
+            return request_codec.decode(data)
+
+        def response_serializer(response):
+            return response_codec.encode(response)
+
+        return request_deserializer, response_serializer

src/dubbo/codec/json_codec/__init__.py

@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from .json_codec_handler import JsonCodec
+
+__all__ = ["JsonCodec"]

src/dubbo/codec/json_codec/json_codec_handler.py

@@ -0,0 +1,50 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from abc import ABC, abstractmethod
+from typing import Any, Type, Protocol, Callable, TypeVar, Generic
+import dubbo
+from functools import wraps
+from pydantic import BaseModel
+from typing import Any, Type, TypeVar, Generic
+from pydantic import BaseModel
+from dubbo.codec.base_codec import Codec
+import orjson
+
+class EncodingFunction(Protocol):
+    def __call__(self, obj: Any) -> bytes: ...
+
+
+class DecodingFunction(Protocol):
+    def __call__(self, data: bytes) -> Any: ...
+
+ModelT = TypeVar('ModelT', bound=BaseModel)
+
+class JsonCodec(Codec, Generic[ModelT]):
+    """JSON codec for Pydantic models using orjson for performance"""
+
+    def __init__(self, model_type: Type[ModelT]):
+        self.model_type = model_type
+
+    def encode(self, data: Any) -> bytes:
+        if isinstance(data, dict):
+            data = self.model_type(**data)
+        elif not isinstance(data, self.model_type):
+            raise TypeError(f"Expected {self.model_type.__name__} or dict, got {type(data).__name__}")
+        return orjson.dumps(data.model_dump())
+
+    def decode(self, data: bytes) -> ModelT:
+        json_data = orjson.loads(data)
+        return self.model_type(**json_data)