# Copyright 1999-2025 Alibaba Group Holding Ltd. # # Licensed 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. import logging import weakref from collections import OrderedDict from typing import Any, Dict, List, Optional, Tuple, Type import msgpack from ...errors import MaxFrameDeprecationError from ...lib.mmh3 import hash from ...utils import no_default from ..core import Placeholder, Serializer, buffered, load_type from .field import Field from .field_type import DictType, ListType, PrimitiveFieldType, TupleType try: from ..deserializer import get_legacy_module_name except ImportError: get_legacy_module_name = lambda x: x logger = logging.getLogger(__name__) _deprecate_log_key = "_SER_DEPRECATE_LOGGED" def _is_field_primitive_compound(field: Field): if field.on_serialize is not None or field.on_deserialize is not None: return False def check_type(field_type): if isinstance(field_type, PrimitiveFieldType): return True if isinstance(field_type, (ListType, TupleType)): if all( check_type(element_type) or element_type is Ellipsis for element_type in field_type._field_types ): return True if isinstance(field_type, DictType): if all( isinstance(element_type, PrimitiveFieldType) or element_type is Ellipsis for element_type in (field_type.key_type, field_type.value_type) ): return True return False return check_type(field.field_type) class SerializableMeta(type): def __new__(mcs, name: str, bases: Tuple[Type], properties: Dict): # All the fields including misc fields. legacy_name_hash = hash( f"{get_legacy_module_name(properties.get('__module__'))}.{name}" ) name_hash = hash( f"{properties.get('__module__')}.{properties.get('__qualname__')}" ) all_fields = dict() # mapping field names to base classes field_to_cls_hash = dict() # mapping legacy name hash to name hashes legacy_to_new_name_hash = {legacy_name_hash: name_hash} for base in bases: if not hasattr(base, "_FIELDS"): continue all_fields.update(base._FIELDS) field_to_cls_hash.update(base._FIELD_TO_NAME_HASH) legacy_to_new_name_hash.update(base._LEGACY_TO_NEW_NAME_HASH) properties_without_fields = {} properties_field_slot_names = [] for k, v in properties.items(): if not isinstance(v, Field): properties_without_fields[k] = v continue field = all_fields.get(k) # record the field for the class being created field_to_cls_hash[k] = name_hash if field is None: properties_field_slot_names.append(k) else: v.name = field.name v.get = field.get v.set = field.set v.__delete__ = field.__delete__ all_fields[k] = v # Make field order deterministic to serialize it as list instead of dict. field_order = list(all_fields) primitive_fields = [] primitive_field_names = set() non_primitive_fields = [] for field_name, v in all_fields.items(): if _is_field_primitive_compound(v): primitive_fields.append(v) primitive_field_names.add(field_name) else: non_primitive_fields.append(v) # count number of fields for every base class cls_to_primitive_field_count = OrderedDict() cls_to_non_primitive_field_count = OrderedDict() for field_name in field_order: cls_hash = field_to_cls_hash[field_name] if field_name in primitive_field_names: cls_to_primitive_field_count[cls_hash] = ( cls_to_primitive_field_count.get(cls_hash, 0) + 1 ) else: cls_to_non_primitive_field_count[cls_hash] = ( cls_to_non_primitive_field_count.get(cls_hash, 0) + 1 ) slots = set(properties.pop("__slots__", set())) slots.update(properties_field_slot_names) properties = properties_without_fields # todo remove this prop when all versions below v1.0.0rc1 is eliminated properties["_LEGACY_NAME_HASH"] = legacy_name_hash properties["_NAME_HASH"] = name_hash properties["_LEGACY_TO_NEW_NAME_HASH"] = legacy_to_new_name_hash properties["_FIELDS"] = all_fields properties["_FIELD_ORDER"] = field_order properties["_FIELD_TO_NAME_HASH"] = field_to_cls_hash properties["_PRIMITIVE_FIELDS"] = primitive_fields properties["_CLS_TO_PRIMITIVE_FIELD_COUNT"] = OrderedDict( cls_to_primitive_field_count ) properties["_NON_PRIMITIVE_FIELDS"] = non_primitive_fields properties["_CLS_TO_NON_PRIMITIVE_FIELD_COUNT"] = OrderedDict( cls_to_non_primitive_field_count ) properties["__slots__"] = tuple(slots) clz = type.__new__(mcs, name, bases, properties) # Bind slot member_descriptor with field. for name in properties_field_slot_names: member_descriptor = getattr(clz, name) field = all_fields[name] field.name = member_descriptor.__name__ field.get = member_descriptor.__get__ field.set = member_descriptor.__set__ field.__delete__ = member_descriptor.__delete__ setattr(clz, name, field) return clz class Serializable(metaclass=SerializableMeta): __slots__ = ("__weakref__",) _cache_primitive_serial = False _ignore_non_existing_keys = False _LEGACY_NAME_HASH: int _NAME_HASH: int _LEGACY_TO_NEW_NAME_HASH: Dict[int, int] _FIELDS: Dict[str, Field] _FIELD_ORDER: List[str] _FIELD_TO_NAME_HASH: Dict[str, int] _PRIMITIVE_FIELDS: List[str] _CLS_TO_PRIMITIVE_FIELD_COUNT: Dict[int, int] _NON_PRIMITIVE_FIELDS: List[str] _CLS_TO_NON_PRIMITIVE_FIELD_COUNT: Dict[int, int] def __init__(self, *args, **kwargs): fields = self._FIELDS field_order = self._FIELD_ORDER assert len(args) <= len(field_order) if args: # pragma: no cover values = dict(zip(field_order, args)) values.update(kwargs) else: values = kwargs for k, v in values.items(): try: fields[k].set(self, v) except KeyError: if not self._ignore_non_existing_keys: raise def __on_deserialize__(self): pass def __repr__(self): values = ", ".join( [ "{}={!r}".format(slot, getattr(self, slot, None)) for slot in self.__slots__ ] ) return "{}({})".format(self.__class__.__name__, values) def copy_to(self, target: "Serializable") -> "Serializable": copied_fields = target._FIELDS for k, field in self._FIELDS.items(): try: # Slightly faster than getattr. value = field.get(self, k) try: copied_fields[k].set(target, value) except KeyError: copied_fields["_" + k].set(target, value) except AttributeError: continue return target def copy(self) -> "Serializable": return self.copy_to(type(self)()) _primitive_serial_cache = weakref.WeakKeyDictionary() class _NoFieldValue: pass _no_field_value = _NoFieldValue() def _to_primitive_placeholder(v: Any) -> Any: if v is _no_field_value or v is no_default: return {} return v def _restore_primitive_placeholder(v: Any) -> Any: if type(v) is dict: if v == {}: return _no_field_value else: return v else: return v class SerializableSerializer(Serializer): """ Leverage DictSerializer to perform serde. """ @classmethod def _log_legacy(cls, context: Dict, key: Any, msg: str, *args, **kwargs): level = kwargs.pop("level", logging.WARNING) try: logged_keys = context[_deprecate_log_key] except KeyError: logged_keys = context[_deprecate_log_key] = set() if key not in logged_keys: logged_keys.add(key) logger.log(level, msg, *args, **kwargs) @classmethod def _get_obj_field_count_key(cls, obj: Serializable, legacy: bool = False): return f"FC_{obj._NAME_HASH if not legacy else obj._LEGACY_NAME_HASH}" @classmethod def _get_field_values(cls, obj: Serializable, fields): values = [] for field in fields: try: value = field.get(obj) if field.on_serialize is not None: value = field.on_serialize(value) except AttributeError: # Most field values are not None, serialize by list is more efficient than dict. value = _no_field_value values.append(value) return values @buffered def serial(self, obj: Serializable, context: Dict): if obj._cache_primitive_serial and obj in _primitive_serial_cache: primitive_vals = _primitive_serial_cache[obj] else: primitive_vals = self._get_field_values(obj, obj._PRIMITIVE_FIELDS) # replace _no_field_value as {} to make them msgpack-serializable primitive_vals = [_to_primitive_placeholder(v) for v in primitive_vals] if obj._cache_primitive_serial: primitive_vals = msgpack.dumps(primitive_vals) _primitive_serial_cache[obj] = primitive_vals compound_vals = self._get_field_values(obj, obj._NON_PRIMITIVE_FIELDS) cls_module = f"{type(obj).__module__}#{type(obj).__qualname__}" field_count_key = self._get_obj_field_count_key(obj) if not self.is_public_data_exist(context, field_count_key): # store field distribution for current Serializable counts = [ list(obj._CLS_TO_PRIMITIVE_FIELD_COUNT.items()), list(obj._CLS_TO_NON_PRIMITIVE_FIELD_COUNT.items()), ] field_count_data = msgpack.dumps(counts) self.put_public_data( context, self._get_obj_field_count_key(obj), field_count_data ) return [cls_module, primitive_vals], [compound_vals], False @staticmethod def _set_field_value(obj: Serializable, field: Field, value): if value is _no_field_value: return if type(value) is Placeholder: if field.on_deserialize is not None: value.callbacks.append( lambda v: field.set(obj, field.on_deserialize(v)) ) else: value.callbacks.append(lambda v: field.set(obj, v)) else: if field.on_deserialize is not None: field.set(obj, field.on_deserialize(value)) else: field.set(obj, value) @classmethod def _prune_server_fields( cls, client_cls_to_field_count: Optional[Dict[int, int]], server_cls_to_field_count: Dict[int, int], server_fields: list, legacy_to_new_hash: Dict[int, int], ) -> list: if set(client_cls_to_field_count.keys()) == set( server_cls_to_field_count.keys() ): return server_fields new_to_legacy_hash = {v: k for k, v in legacy_to_new_hash.items()} ret_server_fields = [] server_pos = 0 for cls_hash, count in server_cls_to_field_count.items(): if ( cls_hash in client_cls_to_field_count or new_to_legacy_hash.get(cls_hash) in client_cls_to_field_count ): ret_server_fields.extend(server_fields[server_pos : server_pos + count]) server_pos += count return ret_server_fields @classmethod def _set_field_values( cls, obj: Serializable, values: List[Any], client_cls_to_field_count: Optional[Dict[int, int]], is_primitive: bool = True, ): obj_class = type(obj) legacy_to_new_hash = obj_class._LEGACY_TO_NEW_NAME_HASH if is_primitive: server_cls_to_field_count = obj_class._CLS_TO_PRIMITIVE_FIELD_COUNT field_def_list = obj_class._PRIMITIVE_FIELDS else: server_cls_to_field_count = obj_class._CLS_TO_NON_PRIMITIVE_FIELD_COUNT field_def_list = obj_class._NON_PRIMITIVE_FIELDS server_fields = cls._prune_server_fields( client_cls_to_field_count, server_cls_to_field_count, field_def_list, legacy_to_new_hash, ) field_num, server_field_num = 0, 0 for cls_hash, count in client_cls_to_field_count.items(): # cut values and fields given field distribution # at client and server end cls_fields = server_fields[server_field_num : field_num + count] cls_values = values[field_num : field_num + count] for field, value in zip(cls_fields, cls_values): if is_primitive: value = _restore_primitive_placeholder(value) if not is_primitive or value is not _no_field_value: cls._set_field_value(obj, field, value) field_num += count try: server_field_num += server_cls_to_field_count[cls_hash] except KeyError: try: server_field_num += server_cls_to_field_count[ legacy_to_new_hash[cls_hash] ] except KeyError: # it is possible that certain type of field does not exist # at server side pass def deserial(self, serialized: List, context: Dict, subs: List) -> Serializable: obj_class_name, primitives = serialized obj_class = load_type(obj_class_name, Serializable) if type(primitives) is not list: primitives = msgpack.loads(primitives) obj = obj_class.__new__(obj_class) field_count_data = self.get_public_data( context, self._get_obj_field_count_key(obj) ) if field_count_data is None: # try using legacy field count key to get counts field_count_data = self.get_public_data( context, self._get_obj_field_count_key(obj, legacy=True) ) if field_count_data is None: self._log_legacy( context, ("MISSING_CLASS", obj_class_name), "Field count info of %s not found in serialized data", obj_class_name, level=logging.ERROR, ) raise MaxFrameDeprecationError( "Failed to deserialize request. Please upgrade your " "MaxFrame client to the latest release." ) else: self._log_legacy( context, ("LEGACY_CLASS", obj_class_name), "Class %s used in legacy client", obj_class_name, ) cls_to_prim_key, cls_to_non_prim_key = msgpack.loads(field_count_data) cls_to_prim_key = dict(cls_to_prim_key) cls_to_non_prim_key = dict(cls_to_non_prim_key) if primitives: self._set_field_values(obj, primitives, cls_to_prim_key, True) if obj_class._NON_PRIMITIVE_FIELDS: self._set_field_values(obj, subs[0], cls_to_non_prim_key, False) obj.__on_deserialize__() return obj class NoFieldValueSerializer(Serializer): def serial(self, obj, context): return [], [], True def deserial(self, serialized, context, subs): return _no_field_value SerializableSerializer.register(Serializable) NoFieldValueSerializer.register(_NoFieldValue)