cnzakii commented on code in PR #51: URL: https://github.com/apache/dubbo-python/pull/51#discussion_r2317631121
########## src/dubbo/client.py: ########## @@ -8,13 +8,15 @@ # # 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, +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an "AS IS" BASIS, Review Comment: Why is it necessary to modify the license? We should keep the license consistent, even down to line break positions. ########## src/dubbo/client.py: ########## @@ -82,82 +91,116 @@ def _initialize(self): self._initialized = True - def unary( + @classmethod Review Comment: This function’s inference logic is too simplistic. * For methods carrying `self`, this should be treated as an invalid case and raise an exception. * It only infers parameter types, but it could also infer default values, whether a parameter is `POSITION_ONLY`, `KEYWORD_ONLY`, etc., which would be very useful for our downstream processing. * The current implementation does not handle streaming scenarios, which makes this inference method unusable in practice. I suggest temporarily removing this inference logic and only allowing users to manually provide input/output definitions. I’ve already implemented a more complete inference function, which I plan to merge later. ########## src/dubbo/codec/dubbo_codec.py: ########## @@ -0,0 +1,160 @@ +# +# 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 typing import Any, Type, Optional, Callable, List, Dict +from dataclasses import dataclass +import inspect + +from dubbo.classes import CodecHelper +from dubbo.codec.json_codec import JsonTransportCodec, JsonTransportEncoder, JsonTransportDecoder + +@dataclass +class ParameterDescriptor: + """Detailed information about a method parameter""" + name: str + annotation: Any + is_required: bool = True + default_value: Any = None + + +@dataclass +class MethodDescriptor: + """Complete method descriptor with all necessary information""" + function: Callable + name: str + parameters: List[ParameterDescriptor] + return_parameter: ParameterDescriptor + documentation: Optional[str] = None + + +class DubboTransportService: + """Enhanced Dubbo transport service with robust type handling""" + + @staticmethod + def create_transport_codec(transport_type: str = 'json', parameter_types: List[Type] = None, Review Comment: There’s no return type hint here, so it’s unclear what this function actually returns. ########## src/dubbo/codec/dubbo_codec.py: ########## @@ -0,0 +1,160 @@ +# +# 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 typing import Any, Type, Optional, Callable, List, Dict +from dataclasses import dataclass +import inspect + +from dubbo.classes import CodecHelper +from dubbo.codec.json_codec import JsonTransportCodec, JsonTransportEncoder, JsonTransportDecoder + +@dataclass +class ParameterDescriptor: + """Detailed information about a method parameter""" + name: str + annotation: Any + is_required: bool = True + default_value: Any = None + + +@dataclass +class MethodDescriptor: + """Complete method descriptor with all necessary information""" + function: Callable + name: str + parameters: List[ParameterDescriptor] + return_parameter: ParameterDescriptor + documentation: Optional[str] = None + + +class DubboTransportService: Review Comment: The name DubboTransportService is `misleading` — it suggests something related to the transport layer, while it actually belongs to the serialization layer. ########## src/dubbo/codec/dubbo_codec.py: ########## @@ -0,0 +1,160 @@ +# +# 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 typing import Any, Type, Optional, Callable, List, Dict +from dataclasses import dataclass +import inspect + +from dubbo.classes import CodecHelper +from dubbo.codec.json_codec import JsonTransportCodec, JsonTransportEncoder, JsonTransportDecoder + +@dataclass +class ParameterDescriptor: + """Detailed information about a method parameter""" + name: str + annotation: Any + is_required: bool = True + default_value: Any = None + + +@dataclass +class MethodDescriptor: + """Complete method descriptor with all necessary information""" + function: Callable + name: str + parameters: List[ParameterDescriptor] + return_parameter: ParameterDescriptor + documentation: Optional[str] = None + + +class DubboTransportService: + """Enhanced Dubbo transport service with robust type handling""" + + @staticmethod + def create_transport_codec(transport_type: str = 'json', parameter_types: List[Type] = None, + return_type: Type = None, **codec_options): + """Create transport codec with enhanced parameter structure""" + if transport_type == 'json': + return JsonTransportCodec( + parameter_types=parameter_types, + return_type=return_type, + **codec_options Review Comment: Why isn’t JSON handled through the plugin mechanism? In Dubbo, anything that is replaceable should be provided as a plugin. ########## src/dubbo/codec/json_codec/json_codec_handler.py: ########## @@ -0,0 +1,322 @@ +# +# 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 typing import Any, Type, List, Union, Dict, TypeVar, Protocol +from datetime import datetime, date, time +from decimal import Decimal +from pathlib import Path +from uuid import UUID +import json + +from .json_type import ( + TypeProviderFactory, SerializationState, + SerializationException, DeserializationException +) + +try: + import orjson + HAS_ORJSON = True +except ImportError: + HAS_ORJSON = False + +try: + import ujson + HAS_UJSON = True +except ImportError: + HAS_UJSON = False + +try: + from pydantic import BaseModel, create_model + HAS_PYDANTIC = True +except ImportError: + HAS_PYDANTIC = False Review Comment: Grouping `json`, `orjson`, `ujson`, and `pydantic` under one module seems inappropriate and conflicts with our plugin principle. Each implementation should ideally be a separate plugin. If splitting `json`, `orjson`, and `ujson` causes too much duplication, consider exposing a shared extension point. `pydantic` should remain separate, since it offers unique features like stronger type inference and flexible default handling. ########## src/dubbo/codec/json_codec/json_codec_handler.py: ########## @@ -0,0 +1,322 @@ +# +# 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 typing import Any, Type, List, Union, Dict, TypeVar, Protocol +from datetime import datetime, date, time +from decimal import Decimal +from pathlib import Path +from uuid import UUID +import json + +from .json_type import ( + TypeProviderFactory, SerializationState, + SerializationException, DeserializationException +) + +try: + import orjson + HAS_ORJSON = True +except ImportError: + HAS_ORJSON = False + +try: + import ujson + HAS_UJSON = True +except ImportError: + HAS_UJSON = False + +try: + from pydantic import BaseModel, create_model + HAS_PYDANTIC = True +except ImportError: + HAS_PYDANTIC = False + + +class EncodingFunction(Protocol): + def __call__(self, obj: Any) -> bytes: ... + + +class DecodingFunction(Protocol): + def __call__(self, data: bytes) -> Any: ... + + +ModelT = TypeVar('ModelT', bound=BaseModel) + + +class CustomJSONEncoder(json.JSONEncoder): + def default(self, obj): + if isinstance(obj, datetime): + return { + "__datetime__": obj.isoformat(), + "__timezone__": str(obj.tzinfo) if obj.tzinfo else None + } + elif isinstance(obj, date): + return {"__date__": obj.isoformat()} + elif isinstance(obj, time): + return {"__time__": obj.isoformat()} + elif isinstance(obj, Decimal): + return {"__decimal__": str(obj)} + elif isinstance(obj, (set, frozenset)): + return { + "__frozenset__" if isinstance(obj, frozenset) else "__set__": list(obj) + } + elif isinstance(obj, UUID): + return {"__uuid__": str(obj)} + elif isinstance(obj, Path): + return {"__path__": str(obj)} + else: + return {"__fallback_string__": str(obj), "__original_type__": type(obj).__name__} + + +class JsonTransportEncoder: + def __init__(self, parameter_types: List[Type] = None, maximum_depth: int = 100, + strict_validation: bool = True, **kwargs): + self.parameter_types = parameter_types or [] + self.maximum_depth = maximum_depth + self.strict_validation = strict_validation + self.type_registry = TypeProviderFactory.create_default_registry() + self.custom_encoder = CustomJSONEncoder(ensure_ascii=False, separators=(',', ':')) + self.single_parameter_mode = len(self.parameter_types) == 1 + self.multiple_parameter_mode = len(self.parameter_types) > 1 + if self.multiple_parameter_mode and HAS_PYDANTIC: + self.parameter_wrapper_model = self._create_parameter_wrapper_model() + + def _create_parameter_wrapper_model(self) -> Type[BaseModel]: + model_fields = {} + for i, param_type in enumerate(self.parameter_types): + model_fields[f"parameter_{i}"] = (param_type, ...) + return create_model('MethodParametersWrapper', **model_fields) + + def register_type_provider(self, provider) -> None: + self.type_registry.register_provider(provider) + + def encode(self, arguments: tuple) -> bytes: + try: + if not arguments: + return self._serialize_to_json_bytes([]) + + if self.single_parameter_mode: + parameter = arguments[0] + serialized_param = self._serialize_with_state(parameter) + if HAS_PYDANTIC and isinstance(parameter, BaseModel): + if hasattr(parameter, 'model_dump'): + return self._serialize_to_json_bytes(parameter.model_dump()) + return self._serialize_to_json_bytes(parameter.dict()) + elif isinstance(parameter, dict): + return self._serialize_to_json_bytes(serialized_param) + else: + return self._serialize_to_json_bytes(serialized_param) + + elif self.multiple_parameter_mode and HAS_PYDANTIC: Review Comment: This handling is too hasty. The input parameters can be either positional or keyword (excluding mixed cases). Since your inference above doesn’t parse the `parameter Kind` and only allows positional arguments, it greatly limits the flexibility for users. ########## src/dubbo/codec/dubbo_codec.py: ########## @@ -0,0 +1,160 @@ +# +# 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 typing import Any, Type, Optional, Callable, List, Dict +from dataclasses import dataclass +import inspect + +from dubbo.classes import CodecHelper +from dubbo.codec.json_codec import JsonTransportCodec, JsonTransportEncoder, JsonTransportDecoder + +@dataclass +class ParameterDescriptor: + """Detailed information about a method parameter""" + name: str + annotation: Any + is_required: bool = True + default_value: Any = None + + +@dataclass +class MethodDescriptor: + """Complete method descriptor with all necessary information""" + function: Callable + name: str + parameters: List[ParameterDescriptor] + return_parameter: ParameterDescriptor + documentation: Optional[str] = None + + +class DubboTransportService: + """Enhanced Dubbo transport service with robust type handling""" + + @staticmethod + def create_transport_codec(transport_type: str = 'json', parameter_types: List[Type] = None, + return_type: Type = None, **codec_options): + """Create transport codec with enhanced parameter structure""" + if transport_type == 'json': + return JsonTransportCodec( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + else: + from dubbo.extension.extension_loader import ExtensionLoader + Codec = CodecHelper.get_class() + codec_class = ExtensionLoader().get_extension(Codec, transport_type) + return codec_class( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + + @staticmethod + def create_encoder_decoder_pair(transport_type: str, parameter_types: List[Type] = None, + return_type: Type = None, **codec_options) -> tuple[any,any]: + """Create separate encoder and decoder instances""" + + if transport_type == 'json': + parameter_encoder = JsonTransportEncoder(parameter_types=parameter_types, **codec_options) + return_decoder = JsonTransportDecoder(target_type=return_type, **codec_options) + return parameter_encoder, return_decoder + else: + from dubbo.extension.extension_loader import ExtensionLoader + Codec = CodecHelper.get_class() + codec_class = ExtensionLoader().get_extension(Codec, transport_type) + + codec_instance = codec_class( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + return codec_instance.get_encoder(), codec_instance.get_decoder() + + @staticmethod + def create_serialization_functions(transport_type: str, parameter_types: List[Type] = None, + return_type: Type = None, **codec_options) -> tuple[Callable, Callable]: + """Create serializer and deserializer functions for RPC (backward compatibility)""" + + parameter_encoder, return_decoder = DubboTransportService.create_encoder_decoder_pair( + transport_type=transport_type, + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + + def serialize_method_parameters(*args) -> bytes: + return parameter_encoder.encode(args) + + def deserialize_method_return(data: bytes): + return return_decoder.decode(data) + + return serialize_method_parameters, deserialize_method_return + Review Comment: Is this compatibility code? ########## src/dubbo/codec/json_codec/json_type.py: ########## @@ -0,0 +1,274 @@ +# +# 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, + List, + Dict, + Set, + Protocol, + runtime_checkable, + Union, +) +from dataclasses import dataclass, fields, is_dataclass, asdict +from datetime import datetime, date, time +from decimal import Decimal +from collections import namedtuple +from pathlib import Path +from uuid import UUID +from enum import Enum +import weakref + +try: + from pydantic import BaseModel + + HAS_PYDANTIC = True +except ImportError: + HAS_PYDANTIC = False + + +class SerializationException(Exception): + """Exception raised during serialization""" + pass + +class DeserializationException(Exception): + """Exception raised during deserialization""" + pass + +class CircularReferenceException(SerializationException): + """Exception raised when circular references are detected""" + pass + +@dataclass(frozen=True) +class SerializationState: + _visited_objects: Set[int] = None + maximum_depth: int = 100 + current_depth: int = 0 + + def __post_init__(self): + if self._visited_objects is None: + object.__setattr__(self, "_visited_objects", set()) + + def validate_circular_reference(self, obj: Any) -> None: + object_id = id(obj) + if object_id in self._visited_objects: + raise CircularReferenceException( + f"Circular reference detected for {type(obj).__name__}" + ) + if self.current_depth >= self.maximum_depth: + raise SerializationException( + f"Maximum serialization depth ({self.maximum_depth}) exceeded" + ) + + def create_child_state(self, obj: Any) -> "SerializationState": + new_visited = self._visited_objects.copy() + new_visited.add(id(obj)) + return SerializationState( + _visited_objects=new_visited, + maximum_depth=self.maximum_depth, + current_depth=self.current_depth + 1, + ) + + +@runtime_checkable +class TypeSerializationProvider(Protocol): + def can_serialize_type(self, obj: Any, obj_type: type) -> bool: ... + + def serialize_to_dict(self, obj: Any, state: SerializationState) -> Any: ... + + +class TypeProviderRegistry: + def __init__(self): + self._type_cache: Dict[type, Optional[TypeSerializationProvider]] = {} + self._providers: List[TypeSerializationProvider] = [] + self._weak_cache = weakref.WeakKeyDictionary() + + def register_provider(self, provider: TypeSerializationProvider) -> None: + self._providers.append(provider) + self._type_cache.clear() + self._weak_cache.clear() + + def find_provider_for_object(self, obj: Any) -> Optional[TypeSerializationProvider]: + obj_type = type(obj) + if obj_type in self._type_cache: + return self._type_cache[obj_type] + provider = None + for p in self._providers: + if p.can_serialize_type(obj, obj_type): + provider = p + break + self._type_cache[obj_type] = provider + return provider + + +class DateTimeSerializationProvider: Review Comment: It does not inherit from `TypeSerializationProvider`. ########## src/dubbo/codec/protobuf_codec/protobuf_codec_handler.py: ########## @@ -0,0 +1,305 @@ +# +# 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 typing import Any, Type, Protocol, Optional +from abc import ABC, abstractmethod +import json +from dataclasses import dataclass + +# Betterproto imports Review Comment: This has the same issue as the JSON part. ########## src/dubbo/client.py: ########## @@ -82,82 +91,116 @@ def _initialize(self): self._initialized = True - def unary( + @classmethod + def _infer_types_from_interface(cls, interface: Callable) -> tuple: + """ + Infer method name, parameter types, and return type from a callable. + """ + try: + type_hints = get_type_hints(interface) + sig = inspect.signature(interface) + method_name = interface.__name__ + params = list(sig.parameters.values()) + + # skip 'self' for bound methods + if params and params[0].name == "self": + params = params[1:] + + param_types = [type_hints.get(p.name, Any) for p in params] + return_type = type_hints.get("return", Any) + + return method_name, param_types, return_type + except Exception: + return interface.__name__, [Any], Any + + def _create_rpc_callable( self, - method_name: str, + rpc_type: 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, + default_method_name: str = "rpc_call", Review Comment: There should never be an implicit `default_method_name`. For RPC methods, inference from the function or user-provided input is acceptable, but defining a custom `rpc_call` should never be allowed. ########## src/dubbo/client.py: ########## @@ -82,82 +91,116 @@ def _initialize(self): self._initialized = True - def unary( + @classmethod + def _infer_types_from_interface(cls, interface: Callable) -> tuple: + """ + Infer method name, parameter types, and return type from a callable. + """ + try: + type_hints = get_type_hints(interface) + sig = inspect.signature(interface) + method_name = interface.__name__ + params = list(sig.parameters.values()) + + # skip 'self' for bound methods + if params and params[0].name == "self": + params = params[1:] + + param_types = [type_hints.get(p.name, Any) for p in params] + return_type = type_hints.get("return", Any) + + return method_name, param_types, return_type + except Exception: + return interface.__name__, [Any], Any + + def _create_rpc_callable( self, - method_name: str, + rpc_type: 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, + default_method_name: str = "rpc_call", ) -> 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 RPC callable with the specified type. + """ + if interface is None and method_name is None: + raise ValueError("Either 'interface' or 'method_name' must be provided") + + # Start with explicit values + m_name = method_name + p_types = params_types + r_type = return_type + + # Infer from interface if needed + if interface: + if p_types is None or r_type is None or m_name is None: + inf_name, inf_params, inf_return = self._infer_types_from_interface( + interface + ) + m_name = m_name or inf_name + p_types = p_types or inf_params + r_type = r_type or inf_return + + # Fallback to default + m_name = m_name or default_method_name + + # Determine serializers + if request_serializer and response_deserializer: + req_ser = request_serializer + res_deser = response_deserializer + else: + req_ser, res_deser = DubboTransportService.create_serialization_functions( + codec or "json", # fallback to json Review Comment: Users don’t always need serialization. For example, when both the RPC input and output are already bytes, serialization is unnecessary. That’s why I think making this an explicit input would be better. ########## src/dubbo/codec/dubbo_codec.py: ########## @@ -0,0 +1,160 @@ +# +# 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 typing import Any, Type, Optional, Callable, List, Dict +from dataclasses import dataclass +import inspect + +from dubbo.classes import CodecHelper +from dubbo.codec.json_codec import JsonTransportCodec, JsonTransportEncoder, JsonTransportDecoder + +@dataclass +class ParameterDescriptor: + """Detailed information about a method parameter""" + name: str + annotation: Any + is_required: bool = True + default_value: Any = None + + +@dataclass +class MethodDescriptor: + """Complete method descriptor with all necessary information""" + function: Callable + name: str + parameters: List[ParameterDescriptor] + return_parameter: ParameterDescriptor + documentation: Optional[str] = None + + +class DubboTransportService: + """Enhanced Dubbo transport service with robust type handling""" + + @staticmethod + def create_transport_codec(transport_type: str = 'json', parameter_types: List[Type] = None, + return_type: Type = None, **codec_options): + """Create transport codec with enhanced parameter structure""" + if transport_type == 'json': + return JsonTransportCodec( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + else: + from dubbo.extension.extension_loader import ExtensionLoader + Codec = CodecHelper.get_class() + codec_class = ExtensionLoader().get_extension(Codec, transport_type) + return codec_class( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + + @staticmethod + def create_encoder_decoder_pair(transport_type: str, parameter_types: List[Type] = None, + return_type: Type = None, **codec_options) -> tuple[any,any]: + """Create separate encoder and decoder instances""" + + if transport_type == 'json': + parameter_encoder = JsonTransportEncoder(parameter_types=parameter_types, **codec_options) + return_decoder = JsonTransportDecoder(target_type=return_type, **codec_options) + return parameter_encoder, return_decoder + else: + from dubbo.extension.extension_loader import ExtensionLoader + Codec = CodecHelper.get_class() + codec_class = ExtensionLoader().get_extension(Codec, transport_type) + + codec_instance = codec_class( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + return codec_instance.get_encoder(), codec_instance.get_decoder() + + @staticmethod + def create_serialization_functions(transport_type: str, parameter_types: List[Type] = None, + return_type: Type = None, **codec_options) -> tuple[Callable, Callable]: + """Create serializer and deserializer functions for RPC (backward compatibility)""" + + parameter_encoder, return_decoder = DubboTransportService.create_encoder_decoder_pair( + transport_type=transport_type, + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + + def serialize_method_parameters(*args) -> bytes: + return parameter_encoder.encode(args) + + def deserialize_method_return(data: bytes): + return return_decoder.decode(data) + + return serialize_method_parameters, deserialize_method_return + + @staticmethod + def create_method_descriptor(func: Callable, method_name: str = None, Review Comment: Can a type hint like `method_name: str = None` actually pass `mypy` checks? I don’t think it should. ########## src/dubbo/codec/json_codec/json_codec_handler.py: ########## @@ -0,0 +1,322 @@ +# +# 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 typing import Any, Type, List, Union, Dict, TypeVar, Protocol +from datetime import datetime, date, time +from decimal import Decimal +from pathlib import Path +from uuid import UUID +import json + +from .json_type import ( + TypeProviderFactory, SerializationState, + SerializationException, DeserializationException +) + +try: + import orjson + HAS_ORJSON = True +except ImportError: + HAS_ORJSON = False + +try: + import ujson + HAS_UJSON = True +except ImportError: + HAS_UJSON = False + +try: + from pydantic import BaseModel, create_model + HAS_PYDANTIC = True +except ImportError: + HAS_PYDANTIC = False + + +class EncodingFunction(Protocol): + def __call__(self, obj: Any) -> bytes: ... + + +class DecodingFunction(Protocol): + def __call__(self, data: bytes) -> Any: ... + + +ModelT = TypeVar('ModelT', bound=BaseModel) + + +class CustomJSONEncoder(json.JSONEncoder): + def default(self, obj): + if isinstance(obj, datetime): + return { + "__datetime__": obj.isoformat(), + "__timezone__": str(obj.tzinfo) if obj.tzinfo else None + } + elif isinstance(obj, date): + return {"__date__": obj.isoformat()} + elif isinstance(obj, time): + return {"__time__": obj.isoformat()} + elif isinstance(obj, Decimal): + return {"__decimal__": str(obj)} + elif isinstance(obj, (set, frozenset)): + return { + "__frozenset__" if isinstance(obj, frozenset) else "__set__": list(obj) + } + elif isinstance(obj, UUID): + return {"__uuid__": str(obj)} + elif isinstance(obj, Path): + return {"__path__": str(obj)} + else: + return {"__fallback_string__": str(obj), "__original_type__": type(obj).__name__} + + +class JsonTransportEncoder: + def __init__(self, parameter_types: List[Type] = None, maximum_depth: int = 100, + strict_validation: bool = True, **kwargs): + self.parameter_types = parameter_types or [] + self.maximum_depth = maximum_depth + self.strict_validation = strict_validation + self.type_registry = TypeProviderFactory.create_default_registry() + self.custom_encoder = CustomJSONEncoder(ensure_ascii=False, separators=(',', ':')) + self.single_parameter_mode = len(self.parameter_types) == 1 + self.multiple_parameter_mode = len(self.parameter_types) > 1 + if self.multiple_parameter_mode and HAS_PYDANTIC: + self.parameter_wrapper_model = self._create_parameter_wrapper_model() + + def _create_parameter_wrapper_model(self) -> Type[BaseModel]: + model_fields = {} + for i, param_type in enumerate(self.parameter_types): + model_fields[f"parameter_{i}"] = (param_type, ...) + return create_model('MethodParametersWrapper', **model_fields) + + def register_type_provider(self, provider) -> None: + self.type_registry.register_provider(provider) + + def encode(self, arguments: tuple) -> bytes: + try: + if not arguments: + return self._serialize_to_json_bytes([]) + + if self.single_parameter_mode: + parameter = arguments[0] + serialized_param = self._serialize_with_state(parameter) + if HAS_PYDANTIC and isinstance(parameter, BaseModel): + if hasattr(parameter, 'model_dump'): + return self._serialize_to_json_bytes(parameter.model_dump()) + return self._serialize_to_json_bytes(parameter.dict()) + elif isinstance(parameter, dict): + return self._serialize_to_json_bytes(serialized_param) + else: + return self._serialize_to_json_bytes(serialized_param) + + elif self.multiple_parameter_mode and HAS_PYDANTIC: + wrapper_data = {f"parameter_{i}": arg for i, arg in enumerate(arguments)} + wrapper_instance = self.parameter_wrapper_model(**wrapper_data) + return self._serialize_to_json_bytes(wrapper_instance.model_dump()) + + else: + serialized_args = [self._serialize_with_state(arg) for arg in arguments] + return self._serialize_to_json_bytes(serialized_args) + + except Exception as e: + raise SerializationException(f"Encoding failed: {e}") from e + + def _serialize_with_state(self, obj: Any) -> Any: + state = SerializationState(maximum_depth=self.maximum_depth) + return self._serialize_recursively(obj, state) Review Comment: Is this method intended to prevent recursion from causing an OOM? If so, is it really necessary? ########## src/dubbo/codec/dubbo_codec.py: ########## @@ -0,0 +1,160 @@ +# +# 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 typing import Any, Type, Optional, Callable, List, Dict +from dataclasses import dataclass +import inspect + +from dubbo.classes import CodecHelper +from dubbo.codec.json_codec import JsonTransportCodec, JsonTransportEncoder, JsonTransportDecoder + +@dataclass +class ParameterDescriptor: + """Detailed information about a method parameter""" + name: str + annotation: Any + is_required: bool = True + default_value: Any = None + + +@dataclass +class MethodDescriptor: + """Complete method descriptor with all necessary information""" + function: Callable + name: str + parameters: List[ParameterDescriptor] + return_parameter: ParameterDescriptor + documentation: Optional[str] = None + + +class DubboTransportService: + """Enhanced Dubbo transport service with robust type handling""" + + @staticmethod + def create_transport_codec(transport_type: str = 'json', parameter_types: List[Type] = None, + return_type: Type = None, **codec_options): + """Create transport codec with enhanced parameter structure""" + if transport_type == 'json': + return JsonTransportCodec( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + else: + from dubbo.extension.extension_loader import ExtensionLoader + Codec = CodecHelper.get_class() + codec_class = ExtensionLoader().get_extension(Codec, transport_type) + return codec_class( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + + @staticmethod + def create_encoder_decoder_pair(transport_type: str, parameter_types: List[Type] = None, + return_type: Type = None, **codec_options) -> tuple[any,any]: + """Create separate encoder and decoder instances""" + + if transport_type == 'json': + parameter_encoder = JsonTransportEncoder(parameter_types=parameter_types, **codec_options) + return_decoder = JsonTransportDecoder(target_type=return_type, **codec_options) + return parameter_encoder, return_decoder + else: + from dubbo.extension.extension_loader import ExtensionLoader + Codec = CodecHelper.get_class() + codec_class = ExtensionLoader().get_extension(Codec, transport_type) + + codec_instance = codec_class( + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + return codec_instance.get_encoder(), codec_instance.get_decoder() + + @staticmethod + def create_serialization_functions(transport_type: str, parameter_types: List[Type] = None, + return_type: Type = None, **codec_options) -> tuple[Callable, Callable]: + """Create serializer and deserializer functions for RPC (backward compatibility)""" + + parameter_encoder, return_decoder = DubboTransportService.create_encoder_decoder_pair( + transport_type=transport_type, + parameter_types=parameter_types, + return_type=return_type, + **codec_options + ) + + def serialize_method_parameters(*args) -> bytes: + return parameter_encoder.encode(args) + + def deserialize_method_return(data: bytes): + return return_decoder.decode(data) + + return serialize_method_parameters, deserialize_method_return + + @staticmethod + def create_method_descriptor(func: Callable, method_name: str = None, + parameter_types: List[Type] = None, return_type: Type = None, + interface: Callable = None) -> MethodDescriptor: + """Create a method descriptor from function and configuration""" + + name = method_name or (interface.__name__ if interface else func.__name__) + sig = inspect.signature(interface if interface else func) + + parameters = [] + resolved_parameter_types = parameter_types or [] + + for i, (param_name, param) in enumerate(sig.parameters.items()): + if param_name == 'self': + continue + + param_index = i - 1 if 'self' in sig.parameters else i + + if param_index < len(resolved_parameter_types): + param_type = resolved_parameter_types[param_index] + elif param.annotation != inspect.Parameter.empty: + param_type = param.annotation + else: + param_type = Any + + is_required = param.default == inspect.Parameter.empty + default_value = param.default if not is_required else None + + parameters.append(ParameterDescriptor( + name=param_name, + annotation=param_type, + is_required=is_required, + default_value=default_value + )) + + if return_type: + resolved_return_type = return_type + elif sig.return_annotation != inspect.Signature.empty: + resolved_return_type = sig.return_annotation + else: + resolved_return_type = Any + + return_parameter = ParameterDescriptor( + name="return_value", + annotation=resolved_return_type + ) + Review Comment: There are still some logical errors in this code. You can remove them based on the suggestions above. ########## src/dubbo/codec/json_codec/json_type.py: ########## @@ -0,0 +1,274 @@ +# +# 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, + List, + Dict, + Set, + Protocol, + runtime_checkable, + Union, +) +from dataclasses import dataclass, fields, is_dataclass, asdict +from datetime import datetime, date, time +from decimal import Decimal +from collections import namedtuple +from pathlib import Path +from uuid import UUID +from enum import Enum +import weakref + +try: + from pydantic import BaseModel + + HAS_PYDANTIC = True +except ImportError: + HAS_PYDANTIC = False + + +class SerializationException(Exception): + """Exception raised during serialization""" + pass + +class DeserializationException(Exception): + """Exception raised during deserialization""" + pass + +class CircularReferenceException(SerializationException): + """Exception raised when circular references are detected""" + pass + +@dataclass(frozen=True) +class SerializationState: + _visited_objects: Set[int] = None + maximum_depth: int = 100 + current_depth: int = 0 + + def __post_init__(self): + if self._visited_objects is None: + object.__setattr__(self, "_visited_objects", set()) + + def validate_circular_reference(self, obj: Any) -> None: + object_id = id(obj) + if object_id in self._visited_objects: + raise CircularReferenceException( + f"Circular reference detected for {type(obj).__name__}" + ) + if self.current_depth >= self.maximum_depth: + raise SerializationException( + f"Maximum serialization depth ({self.maximum_depth}) exceeded" + ) + + def create_child_state(self, obj: Any) -> "SerializationState": + new_visited = self._visited_objects.copy() + new_visited.add(id(obj)) + return SerializationState( + _visited_objects=new_visited, + maximum_depth=self.maximum_depth, + current_depth=self.current_depth + 1, + ) + + +@runtime_checkable +class TypeSerializationProvider(Protocol): + def can_serialize_type(self, obj: Any, obj_type: type) -> bool: ... + + def serialize_to_dict(self, obj: Any, state: SerializationState) -> Any: ... + + +class TypeProviderRegistry: + def __init__(self): + self._type_cache: Dict[type, Optional[TypeSerializationProvider]] = {} + self._providers: List[TypeSerializationProvider] = [] + self._weak_cache = weakref.WeakKeyDictionary() + + def register_provider(self, provider: TypeSerializationProvider) -> None: + self._providers.append(provider) + self._type_cache.clear() + self._weak_cache.clear() + + def find_provider_for_object(self, obj: Any) -> Optional[TypeSerializationProvider]: + obj_type = type(obj) + if obj_type in self._type_cache: + return self._type_cache[obj_type] + provider = None + for p in self._providers: + if p.can_serialize_type(obj, obj_type): + provider = p + break + self._type_cache[obj_type] = provider + return provider Review Comment: Is it really necessary to create a complex `TypeProviderRegistry` and `TypeSerializationProvider` to manage serialization methods? Couldn’t a simple `dict[type, Callable]` suffice instead? ########## src/dubbo/codec/json_codec/json_codec_handler.py: ########## @@ -0,0 +1,322 @@ +# +# 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 typing import Any, Type, List, Union, Dict, TypeVar, Protocol +from datetime import datetime, date, time +from decimal import Decimal +from pathlib import Path +from uuid import UUID +import json + +from .json_type import ( + TypeProviderFactory, SerializationState, + SerializationException, DeserializationException +) + +try: + import orjson + HAS_ORJSON = True +except ImportError: + HAS_ORJSON = False + +try: + import ujson + HAS_UJSON = True +except ImportError: + HAS_UJSON = False + +try: + from pydantic import BaseModel, create_model + HAS_PYDANTIC = True +except ImportError: + HAS_PYDANTIC = False + + +class EncodingFunction(Protocol): + def __call__(self, obj: Any) -> bytes: ... + + +class DecodingFunction(Protocol): + def __call__(self, data: bytes) -> Any: ... + + +ModelT = TypeVar('ModelT', bound=BaseModel) + + +class CustomJSONEncoder(json.JSONEncoder): + def default(self, obj): + if isinstance(obj, datetime): + return { + "__datetime__": obj.isoformat(), + "__timezone__": str(obj.tzinfo) if obj.tzinfo else None + } + elif isinstance(obj, date): + return {"__date__": obj.isoformat()} + elif isinstance(obj, time): + return {"__time__": obj.isoformat()} + elif isinstance(obj, Decimal): + return {"__decimal__": str(obj)} + elif isinstance(obj, (set, frozenset)): + return { + "__frozenset__" if isinstance(obj, frozenset) else "__set__": list(obj) + } + elif isinstance(obj, UUID): + return {"__uuid__": str(obj)} + elif isinstance(obj, Path): + return {"__path__": str(obj)} + else: + return {"__fallback_string__": str(obj), "__original_type__": type(obj).__name__} + + +class JsonTransportEncoder: + def __init__(self, parameter_types: List[Type] = None, maximum_depth: int = 100, + strict_validation: bool = True, **kwargs): + self.parameter_types = parameter_types or [] + self.maximum_depth = maximum_depth + self.strict_validation = strict_validation + self.type_registry = TypeProviderFactory.create_default_registry() + self.custom_encoder = CustomJSONEncoder(ensure_ascii=False, separators=(',', ':')) + self.single_parameter_mode = len(self.parameter_types) == 1 + self.multiple_parameter_mode = len(self.parameter_types) > 1 + if self.multiple_parameter_mode and HAS_PYDANTIC: + self.parameter_wrapper_model = self._create_parameter_wrapper_model() + + def _create_parameter_wrapper_model(self) -> Type[BaseModel]: + model_fields = {} + for i, param_type in enumerate(self.parameter_types): + model_fields[f"parameter_{i}"] = (param_type, ...) + return create_model('MethodParametersWrapper', **model_fields) + + def register_type_provider(self, provider) -> None: + self.type_registry.register_provider(provider) + + def encode(self, arguments: tuple) -> bytes: + try: + if not arguments: + return self._serialize_to_json_bytes([]) + + if self.single_parameter_mode: + parameter = arguments[0] + serialized_param = self._serialize_with_state(parameter) + if HAS_PYDANTIC and isinstance(parameter, BaseModel): + if hasattr(parameter, 'model_dump'): + return self._serialize_to_json_bytes(parameter.model_dump()) + return self._serialize_to_json_bytes(parameter.dict()) + elif isinstance(parameter, dict): + return self._serialize_to_json_bytes(serialized_param) + else: + return self._serialize_to_json_bytes(serialized_param) + + elif self.multiple_parameter_mode and HAS_PYDANTIC: + wrapper_data = {f"parameter_{i}": arg for i, arg in enumerate(arguments)} + wrapper_instance = self.parameter_wrapper_model(**wrapper_data) + return self._serialize_to_json_bytes(wrapper_instance.model_dump()) + + else: + serialized_args = [self._serialize_with_state(arg) for arg in arguments] + return self._serialize_to_json_bytes(serialized_args) + + except Exception as e: + raise SerializationException(f"Encoding failed: {e}") from e + + def _serialize_with_state(self, obj: Any) -> Any: + state = SerializationState(maximum_depth=self.maximum_depth) + return self._serialize_recursively(obj, state) + + def _serialize_recursively(self, obj: Any, state: SerializationState) -> Any: + if obj is None or isinstance(obj, (bool, int, float, str)): + return obj + if isinstance(obj, (list, tuple)): + state.validate_circular_reference(obj) + new_state = state.create_child_state(obj) + return [self._serialize_recursively(item, new_state) for item in obj] + elif isinstance(obj, dict): + state.validate_circular_reference(obj) + new_state = state.create_child_state(obj) + result = {} + for key, value in obj.items(): + if not isinstance(key, str): + if self.strict_validation: + raise SerializationException(f"Dictionary key must be string, got {type(key).__name__}") + key = str(key) + result[key] = self._serialize_recursively(value, new_state) + return result + + provider = self.type_registry.find_provider_for_object(obj) + if provider: + try: + serialized = provider.serialize_to_dict(obj, state) + return self._serialize_recursively(serialized, state) + except Exception as e: + if self.strict_validation: + raise SerializationException(f"Provider failed for {type(obj).__name__}: {e}") from e + return {"__serialization_error__": str(e), "__original_type__": type(obj).__name__} + else: + if self.strict_validation: + raise SerializationException(f"No provider for type {type(obj).__name__}") + return {"__fallback_string__": str(obj), "__original_type__": type(obj).__name__} + + def _serialize_to_json_bytes(self, obj: Any) -> bytes: + if HAS_ORJSON: + try: + return orjson.dumps(obj, default=self._orjson_default_handler) + except TypeError: + pass + if HAS_UJSON: + try: + return ujson.dumps(obj, ensure_ascii=False, default=self._ujson_default_handler).encode('utf-8') + except (TypeError, ValueError): + pass + return self.custom_encoder.encode(obj).encode('utf-8') + + def _orjson_default_handler(self, obj): + if isinstance(obj, datetime): + return { + "__datetime__": obj.isoformat(), + "__timezone__": str(obj.tzinfo) if obj.tzinfo else None + } + elif isinstance(obj, date): + return {"__date__": obj.isoformat()} + elif isinstance(obj, time): + return {"__time__": obj.isoformat()} + elif isinstance(obj, Decimal): + return {"__decimal__": str(obj)} + elif isinstance(obj, (set, frozenset)): + return { + "__frozenset__" if isinstance(obj, frozenset) else "__set__": list(obj) + } + elif isinstance(obj, UUID): + return {"__uuid__": str(obj)} + elif isinstance(obj, Path): + return {"__path__": str(obj)} + else: + return {"__fallback_string__": str(obj), "__original_type__": type(obj).__name__} + + def _ujson_default_handler(self, obj): + return self._orjson_default_handler(obj) + Review Comment: These should each be separate implementations of the JSON extension, rather than being coupled here. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
