gsoc pull request

Author: aditya0yadav

src/dubbo/classes.py

@@ -13,10 +13,12 @@
 # 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 abc
 import threading
-from typing import Any, Callable, Optional, Union
-
+from typing import Any, Callable, Optional, Union,Type
+from abc import ABC, abstractmethod
+from pydantic import BaseModel
 from dubbo.types import DeserializingFunction, RpcType, RpcTypes, SerializingFunction
 
 __all__ = [
@@ -244,3 +246,21 @@ class ReadWriteStream(ReadStream, WriteStream, abc.ABC):
     """
 
     pass
+
+
+class Codec(ABC):
+    def __init__(self, model_type: Type[BaseModel] = None, **kwargs):
+        self.model_type = model_type
+
+    @abstractmethod
+    def encode(self, data: Any) -> bytes:
+        pass
+
+    @abstractmethod
+    def decode(self, data: bytes) -> Any:
+        pass
+
+class CodecHelper:
+    @staticmethod
+    def get_class():
+        return Codec

src/dubbo/client.py

@@ -13,8 +13,9 @@
 # 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
+from typing import Optional, Callable, List, Type, Union, Any
 
 from dubbo.bootstrap import Dubbo
 from dubbo.classes import MethodDescriptor
@@ -31,6 +32,7 @@
     SerializingFunction,
 )
 from dubbo.url import URL
+from dubbo.codec import DubboTransportService
 
 __all__ = ["Client"]
 
@@ -84,68 +86,274 @@ def _initialize(self):
 
     def unary(
         self,
-        method_name: str,
+        interface: Optional[Callable] = None,
+        method_name: Optional[str] = None,
+        params_types: Optional[List[Type]] = None,
+        return_type: Optional[Type] = None,
+        codec: Optional[str] = None,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
     ) -> RpcCallable:
-        return self._callable(
-            MethodDescriptor(
-                method_name=method_name,
-                arg_serialization=(request_serializer, None),
-                return_serialization=(None, response_deserializer),
-                rpc_type=RpcTypes.UNARY.value,
+        """
+        Create unary RPC call.
+
+        Supports both automatic mode (via interface) and manual mode (via method_name + params_types + return_type + codec).
+        """
+
+        # Validate
+        if interface is None and method_name is None:
+            raise ValueError("Either 'interface' or 'method_name' must be provided")
+
+        # Determine the actual method name to call
+        actual_method_name = method_name or (interface.__name__ if interface else "unary")
+        
+        # Build method descriptor (automatic or manual)
+        if interface:
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=interface,
+                method_name=actual_method_name,
+                parameter_types=params_types,
+                return_type=return_type,
+                interface=interface,
             )
+        else:
+            # Manual mode fallback: use dummy function for descriptor creation
+            def dummy(): pass
+
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=dummy,
+                method_name=actual_method_name,
+                parameter_types=params_types or [],
+                return_type=return_type or Any,
+            )
+
+        # Determine serializers if not provided
+        if request_serializer and response_deserializer:
+            final_request_serializer = request_serializer
+            final_response_deserializer = response_deserializer
+        else:
+            # Use DubboTransportService to generate serialization functions
+            final_request_serializer, final_response_deserializer = DubboTransportService.create_serialization_functions(
+                transport_type=codec or "json",
+                parameter_types=[p.annotation for p in method_desc.parameters],
+                return_type=method_desc.return_parameter.annotation,
+            )
+
+        # Create the proper MethodDescriptor for the RPC call
+        # This should match the structure expected by your RpcCallableFactory
+        rpc_method_descriptor = MethodDescriptor(
+            method_name=actual_method_name,
+            arg_serialization=(final_request_serializer, None),  # (serializer, deserializer) for arguments
+            return_serialization=(None, final_response_deserializer),  # (serializer, deserializer) for return value
+            rpc_type=RpcTypes.UNARY.value,
         )
 
+        # Create and return the RpcCallable
+        return self._callable(rpc_method_descriptor)
+
     def client_stream(
         self,
-        method_name: str,
+        interface: Optional[Callable] = None,
+        method_name: Optional[str] = None,
+        params_types: Optional[List[Type]] = None,
+        return_type: Optional[Type] = None,
+        codec: Optional[str] = None,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
     ) -> RpcCallable:
-        return self._callable(
-            MethodDescriptor(
-                method_name=method_name,
-                arg_serialization=(request_serializer, None),
-                return_serialization=(None, response_deserializer),
-                rpc_type=RpcTypes.CLIENT_STREAM.value,
+        """
+        Create client streaming RPC call.
+
+        Supports both automatic mode (via interface) and manual mode (via method_name + params_types + return_type + codec).
+        """
+
+        # Validate
+        if interface is None and method_name is None:
+            raise ValueError("Either 'interface' or 'method_name' must be provided")
+
+        # Determine the actual method name to call
+        actual_method_name = method_name or (interface.__name__ if interface else "client_stream")
+        
+        # Build method descriptor (automatic or manual)
+        if interface:
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=interface,
+                method_name=actual_method_name,
+                parameter_types=params_types,
+                return_type=return_type,
+                interface=interface,
             )
+        else:
+            # Manual mode fallback: use dummy function for descriptor creation
+            def dummy(): pass
+
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=dummy,
+                method_name=actual_method_name,
+                parameter_types=params_types or [],
+                return_type=return_type or Any,
+            )
+
+        # Determine serializers if not provided
+        if request_serializer and response_deserializer:
+            final_request_serializer = request_serializer
+            final_response_deserializer = response_deserializer
+        else:
+            # Use DubboTransportService to generate serialization functions
+            final_request_serializer, final_response_deserializer = DubboTransportService.create_serialization_functions(
+                transport_type=codec or "json",
+                parameter_types=[p.annotation for p in method_desc.parameters],
+                return_type=method_desc.return_parameter.annotation,
+            )
+
+        # Create the proper MethodDescriptor for the RPC call
+        # This should match the structure expected by your RpcCallableFactory
+        rpc_method_descriptor = MethodDescriptor(
+            method_name=actual_method_name,
+            arg_serialization=(final_request_serializer, None),  # (serializer, deserializer) for arguments
+            return_serialization=(None, final_response_deserializer),  # (serializer, deserializer) for return value
+            rpc_type=RpcTypes.CLIENT_STREAM.value,
         )
 
+        # Create and return the RpcCallable
+        return self._callable(rpc_method_descriptor)
+
     def server_stream(
         self,
-        method_name: str,
+        interface: Optional[Callable] = None,
+        method_name: Optional[str] = None,
+        params_types: Optional[List[Type]] = None,
+        return_type: Optional[Type] = None,
+        codec: Optional[str] = None,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
     ) -> RpcCallable:
-        return self._callable(
-            MethodDescriptor(
-                method_name=method_name,
-                arg_serialization=(request_serializer, None),
-                return_serialization=(None, response_deserializer),
-                rpc_type=RpcTypes.SERVER_STREAM.value,
+        """
+        Create server streaming RPC call.
+
+        Supports both automatic mode (via interface) and manual mode (via method_name + params_types + return_type + codec).
+        """
+
+        # Validate
+        if interface is None and method_name is None:
+            raise ValueError("Either 'interface' or 'method_name' must be provided")
+
+        # Determine the actual method name to call
+        actual_method_name = method_name or (interface.__name__ if interface else "server_stream")
+        
+        # Build method descriptor (automatic or manual)
+        if interface:
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=interface,
+                method_name=actual_method_name,
+                parameter_types=params_types,
+                return_type=return_type,
+                interface=interface,
+            )
+        else:
+            # Manual mode fallback: use dummy function for descriptor creation
+            def dummy(): pass
+
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=dummy,
+                method_name=actual_method_name,
+                parameter_types=params_types or [],
+                return_type=return_type or Any,
+            )
+
+        # Determine serializers if not provided
+        if request_serializer and response_deserializer:
+            final_request_serializer = request_serializer
+            final_response_deserializer = response_deserializer
+        else:
+            # Use DubboTransportService to generate serialization functions
+            final_request_serializer, final_response_deserializer = DubboTransportService.create_serialization_functions(
+                transport_type=codec or "json",
+                parameter_types=[p.annotation for p in method_desc.parameters],
+                return_type=method_desc.return_parameter.annotation,
             )
+
+        # Create the proper MethodDescriptor for the RPC call
+        # This should match the structure expected by your RpcCallableFactory
+        rpc_method_descriptor = MethodDescriptor(
+            method_name=actual_method_name,
+            arg_serialization=(final_request_serializer, None),  # (serializer, deserializer) for arguments
+            return_serialization=(None, final_response_deserializer),  # (serializer, deserializer) for return value
+            rpc_type=RpcTypes.SERVER_STREAM.value,
         )
 
+        # Create and return the RpcCallable
+        return self._callable(rpc_method_descriptor)
+
     def bi_stream(
         self,
-        method_name: str,
+        interface: Optional[Callable] = None,
+        method_name: Optional[str] = None,
+        params_types: Optional[List[Type]] = None,
+        return_type: Optional[Type] = None,
+        codec: Optional[str] = None,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
     ) -> RpcCallable:
-        # create method descriptor
-        return self._callable(
-            MethodDescriptor(
-                method_name=method_name,
-                arg_serialization=(request_serializer, None),
-                return_serialization=(None, response_deserializer),
-                rpc_type=RpcTypes.BI_STREAM.value,
+        """
+        Create bidirectional streaming RPC call.
+
+        Supports both automatic mode (via interface) and manual mode (via method_name + params_types + return_type + codec).
+        """
+
+        # Validate
+        if interface is None and method_name is None:
+            raise ValueError("Either 'interface' or 'method_name' must be provided")
+
+        # Determine the actual method name to call
+        actual_method_name = method_name or (interface.__name__ if interface else "bi_stream")
+        
+        # Build method descriptor (automatic or manual)
+        if interface:
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=interface,
+                method_name=actual_method_name,
+                parameter_types=params_types,
+                return_type=return_type,
+                interface=interface,
+            )
+        else:
+            # Manual mode fallback: use dummy function for descriptor creation
+            def dummy(): pass
+
+            method_desc = DubboTransportService.create_method_descriptor(
+                func=dummy,
+                method_name=actual_method_name,
+                parameter_types=params_types or [],
+                return_type=return_type or Any,
             )
+
+        # Determine serializers if not provided
+        if request_serializer and response_deserializer:
+            final_request_serializer = request_serializer
+            final_response_deserializer = response_deserializer
+        else:
+            # Use DubboTransportService to generate serialization functions
+            final_request_serializer, final_response_deserializer = DubboTransportService.create_serialization_functions(
+                transport_type=codec or "json",
+                parameter_types=[p.annotation for p in method_desc.parameters],
+                return_type=method_desc.return_parameter.annotation,
+            )
+
+
+        rpc_method_descriptor = MethodDescriptor(
+            method_name=actual_method_name,
+            arg_serialization=(final_request_serializer, None),  
+            return_serialization=(None, final_response_deserializer),  
+            rpc_type=RpcTypes.BI_STREAM.value,
         )
 
+        # Create and return the RpcCallable
+        return self._callable(rpc_method_descriptor)
+
     def _callable(self, method_descriptor: MethodDescriptor) -> RpcCallable:
         """
-        Generate a proxy for the given method
+        Generate a proxy for the given method.
         :param method_descriptor: The method descriptor.
         :return: The proxy.
         :rtype: RpcCallable
@@ -160,4 +368,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,19 @@
+#
+# 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 DubboTransportService
+
+__all__ = ['DubboTransportService']