# 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 __future__ import annotations

import dataclasses

import datetime

import decimal

import enum

import inspect

import logging

import os

import reprlib

import struct

import sys

import typing

from typing import List, Dict

from pyfory.annotation import (

ArrayMeta,

)

from pyfory.lib.mmh3 import hash_buffer

from pyfory.policy import DEFAULT_POLICY

from pyfory.types import (

TypeId,

is_primitive_array_type,

get_primitive_type_size,

is_polymorphic_type,

is_primitive_type,

is_union_type,

)

from pyfory.type_util import (

TypeVisitor,

_get_args,

_get_origin,

infer_field,

infer_field_types,

get_homogeneous_tuple_elem_type,

is_subclass,

get_type_hints,

normalize_fory_type,

scalar_type_id,

unwrap_array,

unwrap_optional,

unwrap_ref,

)

from pyfory.serialization import Buffer

from pyfory.serialization import ENABLE_FORY_CYTHON_SERIALIZATION

from pyfory.error import TypeNotCompatibleError

from pyfory.resolver import NULL_FLAG, NOT_NULL_VALUE_FLAG

from pyfory.field import (

ForyFieldMeta,

extract_field_meta,

validate_field_metas,

)

from pyfory import (

Serializer,

BooleanSerializer,

ByteSerializer,

Int16Serializer,

Int32Serializer,

Int64Serializer,

Float32Serializer,

Float64Serializer,

StringSerializer,

)

logger = logging.getLogger(__name__)

_REFERENCE_BYTES = struct.calcsize("P")

# Lower-bound shallow owner costs for retained Python struct shapes. Normal objects retain an

# instance dict for field storage; slotted objects store field references in object slots.

_SLOTTED_STRUCT_OWNER_BYTES = 4 * _REFERENCE_BYTES

_DICT_BACKED_STRUCT_OWNER_BYTES = 4 * _REFERENCE_BYTES

_INSTANCE_DICT_OWNER_BYTES = 8 * _REFERENCE_BYTES

_MISSING_DEFAULT_INT_TYPES = {

int,

TypeId.INT8,

TypeId.INT16,

TypeId.VARINT32,

TypeId.INT32,

TypeId.VARINT64,

TypeId.INT64,

TypeId.TAGGED_INT64,

TypeId.UINT8,

TypeId.UINT16,

TypeId.VAR_UINT32,

TypeId.UINT32,

TypeId.VAR_UINT64,

TypeId.UINT64,

TypeId.TAGGED_UINT64,

}

_MISSING_DEFAULT_FLOAT_TYPES = {

float,

TypeId.FLOAT16,

TypeId.BFLOAT16,

TypeId.FLOAT32,

TypeId.FLOAT64,

}

_BASIC_FIELD_SERIALIZERS = (

BooleanSerializer,

ByteSerializer,

Int16Serializer,

Int32Serializer,

Int64Serializer,

Float32Serializer,

Float64Serializer,

StringSerializer,

)

class UnknownStruct:

"""Framework-owned value for a checked remote Struct without a local class."""

__slots__ = ("type_def", "_values")

def __init__(self, type_def):

self.type_def = type_def

self._values = {}

def __getitem__(self, key):

return self._values[key]

def get(self, key, default=None):

return self._values.get(key, default)

def items(self):

return self._values.items()

def keys(self):

return self._values.keys()

def values(self):

return self._values.values()

def __iter__(self):

return iter(self._values)

def __len__(self):

return len(self._values)

def __contains__(self, key):

return key in self._values

@reprlib.recursive_repr()

def __repr__(self):

type_def = self.type_def

namespace = getattr(type_def, "namespace", "")

typename = getattr(type_def, "typename", "")

name = f"{namespace}.{typename}" if namespace else typename

return f"UnknownStruct({name!r}, {self._values!r})"

@dataclasses.dataclass

class FieldInfo:

"""Pre-computed field information for serialization."""

# Identity

name: str # Field name (snake_case)

index: int # Field index in the serialization order

type_hint: type # Type annotation

# Fory metadata (from pyfory.field()) - used for hash computation

tag_id: int # -1 is the internal no-ID sentinel, >=0 = use tag ID

nullable: bool # Effective nullable flag (considers Optional[T])

ref: bool # Field-level ref setting (for hash computation)

dynamic: bool # Whether type info is written for this field

# Runtime flags (combines field metadata with global Fory config)

runtime_ref_tracking: bool # Actual ref tracking: field.ref AND fory.track_ref

# Derived info

type_id: int # Fory TypeId

serializer: Serializer # Field serializer

unwrapped_type: type # Type with Optional unwrapped

field_type: object = None # Recursive TypeDef schema node for this field

assign: bool = True # Whether a remote field value may be assigned locally

validation_field_type: object = None # Local schema used for compatible-read validation

class UnknownStructSerializer(Serializer):

"""Schema-driven serializer for the fixed :class:`UnknownStruct` owner."""

__slots__ = (

"type_def",

"_field_names",

"_field_types",

"_field_serializers",

"_basic_fields",

"_dynamic_fields",

"_serializers_by_type_def",

"_graph_memory_bytes",

)

def __init__(self, type_resolver, type_def=None):

super().__init__(type_resolver, UnknownStruct)

self.type_def = type_def

self._field_names = ()

self._field_types = ()

self._field_serializers = ()

self._basic_fields = ()

self._dynamic_fields = ()

self._serializers_by_type_def = {} if type_def is None else None

self._graph_memory_bytes = 0

if type_def is None:

return

self._validate_type_def(type_def)

self._field_names = tuple(field.name for field in type_def.fields)

self._field_types = tuple(field.field_type for field in type_def.fields)

serializers = tuple(field_type.create_serializer(type_resolver, None) for field_type in self._field_types)

for field_type, serializer in zip(self._field_types, serializers):

if serializer is None and not is_polymorphic_type(field_type.type_id):

raise TypeNotCompatibleError(f"Missing compatible field codec for type id {field_type.type_id}")

self._field_serializers = serializers

self._dynamic_fields = tuple(is_polymorphic_type(field_type.type_id) for field_type in self._field_types)

self._basic_fields = tuple(

not dynamic and isinstance(serializer, _BASIC_FIELD_SERIALIZERS) for dynamic, serializer in zip(self._dynamic_fields, serializers)

)

self.read_data_always_advances = any(

field_type.is_nullable or field_type.is_tracking_ref or dynamic or (serializer is not None and serializer.read_data_always_advances)

for field_type, serializer, dynamic in zip(self._field_types, serializers, self._dynamic_fields)

)

field_count = len(self._field_names)

self._graph_memory_bytes = (

_SLOTTED_STRUCT_OWNER_BYTES + 2 * _REFERENCE_BYTES + _INSTANCE_DICT_OWNER_BYTES + field_count * 2 * _REFERENCE_BYTES

)

@staticmethod

def _validate_type_def(type_def):

from pyfory.meta.typedef import TypeDef, is_struct_typedef_kind

if type(type_def) is not TypeDef:

raise TypeError("UnknownStruct requires a framework TypeDef")

if not is_struct_typedef_kind(type_def.type_id) or not TypeId.is_type_share_meta(type_def.type_id):

raise TypeError("UnknownStruct requires a shared Struct TypeDef")

if type(type_def.encoded) is not bytes or len(type_def.encoded) < 8:

raise TypeError("UnknownStruct requires an encoded TypeDef")

def _serializer_for(self, type_def):

self._validate_type_def(type_def)

serializer = self._serializers_by_type_def.get(type_def.encoded)

if serializer is None:

serializer = UnknownStructSerializer(self.type_resolver, type_def)

self._serializers_by_type_def[type_def.encoded] = serializer

return serializer

def _check_value(self, value):

if type(value) is not UnknownStruct:

raise TypeError("UnknownStructSerializer requires an UnknownStruct value")

if value.type_def.encoded != self.type_def.encoded:

raise TypeError("UnknownStruct value does not match serializer TypeDef")

values = value._values

if type(values) is not dict or len(values) != len(self._field_names):

raise TypeError("UnknownStruct fields do not match its TypeDef")

for name in self._field_names:

if name not in values:

raise TypeError(f"UnknownStruct field {name!r} is missing")

return values

def write(self, write_context, value):

if self.type_def is None:

if type(value) is not UnknownStruct:

raise TypeError("UnknownStructSerializer requires an UnknownStruct value")

serializer = self._serializer_for(value.type_def)

values = serializer._check_value(value)

self.type_resolver._write_unknown_struct_type_info(write_context, serializer.type_def)

serializer._write_fields(write_context, values)

return

values = self._check_value(value)

self._write_fields(write_context, values)

def _write_fields(self, write_context, values):

for index, name in enumerate(self._field_names):

field_type = self._field_types[index]

serializer = self._field_serializers[index]

value = values[name]

dynamic = self._dynamic_fields[index]

if self._basic_fields[index]:

if field_type.is_nullable:

if value is None:

write_context.write_int8(NULL_FLAG)

continue

write_context.write_int8(NOT_NULL_VALUE_FLAG)

serializer.write(write_context, value)

elif field_type.is_tracking_ref:

write_context.write_ref(value, serializer=None if dynamic else serializer)

else:

if field_type.is_nullable:

if value is None:

write_context.write_int8(NULL_FLAG)

continue

write_context.write_int8(NOT_NULL_VALUE_FLAG)

write_context.write_no_ref(value, serializer=None if dynamic else serializer)

write_context.try_flush()

def read(self, read_context):

if self.type_def is None:

raise TypeError("Generic UnknownStruct serializer cannot read a value body")

owner = UnknownStruct(self.type_def)

read_context.reserve_graph_memory(self._graph_memory_bytes)

# Publish the exact final owner before reading children. Ordinary

# RefFlag values then resolve cycles without alternate reference state.

read_context.reference(owner)

values = owner._values

for index, name in enumerate(self._field_names):

field_type = self._field_types[index]

serializer = self._field_serializers[index]

dynamic = self._dynamic_fields[index]

if self._basic_fields[index]:

if field_type.is_nullable and read_context.read_int8() == NULL_FLAG:

value = None

else:

value = serializer.read(read_context)

elif field_type.is_tracking_ref:

value = read_context.read_ref(serializer=None if dynamic else serializer)

elif field_type.is_nullable and read_context.read_int8() == NULL_FLAG:

value = None

else:

value = read_context.read_no_ref(serializer=None if dynamic else serializer)

values[name] = value

return owner

def _is_abstract_type(type_hint: type) -> bool:

"""Check if a type is abstract (has abstract methods or is ABC subclass)."""

if type_hint is None:

return False

try:

# Check if it's an abstract class using inspect.isabstract

return inspect.isabstract(type_hint)

except TypeError:

# Not a class (e.g., generic type)

return False

def _is_dynamic_nullable_default(type_hint: type) -> bool:

if type_hint is typing.Any:

return True

origin = _get_origin(type_hint)

args = _get_args(type_hint)

if origin in (list, dict, set, tuple) and not args:

return True

return type_hint in (list, dict, set, tuple, typing.List, typing.Dict, typing.Set, typing.Tuple)

def _default_field_meta(type_hint: type, field_nullable: bool = False, xlang: bool = False) -> ForyFieldMeta:

"""Returns default field metadata for fields without pyfory.field().

A field is considered nullable if:

1. It's Optional[T], OR

2. Global field_nullable is True, OR

3. In Python-native mode, it is a dynamic field such as Any or a bare collection alias.

Xlang fields are non-nullable by default so schema fingerprints and payload flags stay aligned

with other runtimes unless users opt into nullability explicitly.

For ref, defaults to False to preserve original serialization behavior.

Non-nullable complex fields use write_no_ref (no ref header in buffer).

Users can explicitly set ref=True in pyfory.field() to enable ref tracking.

For dynamic, defaults to None (auto-detect):

- Abstract classes: always True (type info must be written)

- Concrete types use type-id based dynamic detection

"""

unwrapped_type, is_optional = unwrap_optional(type_hint)

nullable = is_optional or field_nullable or (not xlang and _is_dynamic_nullable_default(unwrapped_type))

# Default ref=False to preserve original serialization behavior where non-nullable

# fields use write_no_ref. Users can explicitly set ref=True in pyfory.field()

# to enable per-field ref tracking when fory.track_ref is enabled.

# Default dynamic=None for auto-detection based on type and mode

return ForyFieldMeta(id=-1, nullable=nullable, ref=False, ignore=False, dynamic=None)

def _extract_field_infos(

fory,

clz: type,

type_hints: dict,

) -> tuple[list[FieldInfo], dict[str, ForyFieldMeta]]:

"""

Extract FieldInfo list from a dataclass.

This handles:

- Extracting field metadata from pyfory.field() annotations

- Filtering out ignored fields

- Computing effective nullable based on Optional[T]

- Computing runtime ref tracking based on global config

- Inheritance: parent fields first, subclass fields override parent fields

Returns:

Tuple of (field_infos, field_metas) where field_metas maps field name to ForyFieldMeta

"""

if not dataclasses.is_dataclass(clz):

# Non-dataclass registration uses the runtime type inspection path.

return [], {}

# Collect fields from class hierarchy (parent first, child last)

# Child fields override parent fields with same name

all_fields: Dict[str, dataclasses.Field] = {}

for klass in clz.__mro__[::-1]: # Reverse MRO: base classes first

if dataclasses.is_dataclass(klass) and klass is not clz:

for f in dataclasses.fields(klass):

all_fields[f.name] = f

# Add current class fields (override parent)

for f in dataclasses.fields(clz):

all_fields[f.name] = f

# Extract field metas and filter ignored fields

field_metas: Dict[str, ForyFieldMeta] = {}

active_fields: List[tuple] = []

# Check if fory has field_nullable global setting

global_field_nullable = getattr(fory, "field_nullable", False)

for field_name, dc_field in all_fields.items():

meta = extract_field_meta(dc_field)

if meta is None:

# Field without pyfory.field() - use defaults

# Auto-detect Optional[T] for nullable, also respect global field_nullable

field_type = type_hints.get(field_name, typing.Any)

meta = _default_field_meta(field_type, global_field_nullable, getattr(fory, "xlang", False))

field_metas[field_name] = meta

if not meta.ignore:

active_fields.append((field_name, dc_field))

# Validate field metas

validate_field_metas(clz, field_metas, type_hints)

# Build FieldInfo list

field_infos: List[FieldInfo] = []

from pyfory.meta.typedef import build_field_type_with_ref

visitor = StructTypeIdVisitor(fory, clz)

field_types = infer_field_types(clz, field_nullable=global_field_nullable)

global_ref_tracking = fory.track_ref

for index, (field_name, dc_field) in enumerate(active_fields):

meta = field_metas[field_name]

type_hint = type_hints.get(field_name, typing.Any)

unwrapped_type, is_optional = unwrap_optional(type_hint)

# Optional[T] should always be nullable regardless of explicit meta.

effective_nullable = meta.nullable or is_optional

# Compute runtime ref tracking: field.ref AND global config

runtime_ref = meta.ref and global_ref_tracking

field_type = build_field_type_with_ref(

fory,

field_name,

unwrapped_type,

visitor,

effective_nullable,

runtime_ref,

)

array_meta = unwrap_array(unwrapped_type)

if array_meta is not None:

serializer = StructFieldSerializerVisitor(fory).visit_array(

field_name,

array_meta.element_type,

array_meta.carrier,

)

else:

serializer = field_type.create_serializer(fory, field_types.get(field_name, None))

type_id = field_type.type_id

# Compute effective dynamic based on type.

# - Abstract classes: always True (type info must be written)

# - Compatible ref-tracked struct fields carry type info for xlang

# parity with other runtimes

# - If explicitly set (not None): use that value for non-ref fields

# - Otherwise: write type info for polymorphic types that are not registered by id

is_abstract = _is_abstract_type(unwrapped_type)

if fory.compatible and runtime_ref and is_polymorphic_type(type_id):

effective_dynamic = True

elif is_abstract:

# Abstract classes always need type info

effective_dynamic = True

elif meta.dynamic is not None:

# Explicit configuration takes precedence

effective_dynamic = meta.dynamic

else:

# Registered-by-id types have stable serializers, so no per-field type info is needed.

effective_dynamic = is_polymorphic_type(type_id) and not fory.is_registered_by_id(unwrapped_type)

field_info = FieldInfo(

name=field_name,

index=index,

type_hint=type_hint,

tag_id=meta.id,

nullable=effective_nullable,

ref=meta.ref,

dynamic=effective_dynamic,

runtime_ref_tracking=runtime_ref,

type_id=type_id,

serializer=serializer,

unwrapped_type=unwrapped_type,

field_type=field_type,

)

field_infos.append(field_info)

return field_infos, field_metas

def resolve_missing_field_default(

dc_field: dataclasses.Field,

type_resolver,

type_hints: dict[str, typing.Any],

) -> typing.Callable[[], typing.Any]:

type_hint = type_hints.get(dc_field.name, typing.Any)

unwrapped_type, is_optional = unwrap_optional(type_hint)

meta = extract_field_meta(dc_field)

effective_nullable = (meta.nullable if meta is not None else type_resolver.field_nullable) or is_optional

if dc_field.default is not dataclasses.MISSING:

default_value = dc_field.default

if default_value is None and not effective_nullable and is_subclass(unwrapped_type, enum.Enum):

members = tuple(unwrapped_type)

if members:

default_value = members[0]

return lambda value=default_value: value

if dc_field.default_factory is not dataclasses.MISSING:

return dc_field.default_factory

if not effective_nullable:

unwrapped_type = normalize_fory_type(unwrapped_type)

origin = typing.get_origin(unwrapped_type) if hasattr(typing, "get_origin") else getattr(unwrapped_type, "__origin__", None)

origin = origin or unwrapped_type

if is_subclass(unwrapped_type, enum.Enum):

members = tuple(unwrapped_type)

if members:

default_value = members[0]

return lambda value=default_value: value

if origin is list or origin == typing.List:

return lambda: []

if origin is set or origin == typing.Set:

return lambda: set()

if origin is dict or origin == typing.Dict:

return lambda: {}

if unwrapped_type is bool:

return lambda: False

if unwrapped_type in _MISSING_DEFAULT_INT_TYPES:

return lambda: 0

if unwrapped_type in _MISSING_DEFAULT_FLOAT_TYPES:

return lambda: 0.0

if unwrapped_type is str:

return lambda: ""

if unwrapped_type is bytes:

return lambda: b""

return lambda: None

def _resolve_missing_field_default(dc_field, type_resolver, type_hints):

return resolve_missing_field_default(dc_field, type_resolver, type_hints)

def build_default_values_factory(type_resolver, type_hints, dc_fields=()):

return {dc_field.name: _resolve_missing_field_default(dc_field, type_resolver, type_hints) for dc_field in dc_fields}

class DataClassSerializer(Serializer):

_BASIC_SERIALIZERS = _BASIC_FIELD_SERIALIZERS

def __init__(

self,

type_resolver,

clz: type,

field_names: List[str] = None,

serializers: List[Serializer] = None,

nullable_fields: Dict[str, bool] = None,

dynamic_fields: Dict[str, bool] = None,

ref_fields: Dict[str, bool] = None,

field_infos: List[FieldInfo] = None,

fields_from_typedef: bool = False,

):

super().__init__(type_resolver, clz)

self._type_hints = get_type_hints(clz)

self._has_slots = hasattr(clz, "__slots__")

self._fields_from_typedef = fields_from_typedef or (field_names is not None and serializers is not None)

if field_infos is not None:

self._field_infos = list(field_infos)

self._field_names = [fi.name for fi in self._field_infos]

self._serializers = [fi.serializer for fi in self._field_infos]

self._nullable_fields = {fi.name: fi.nullable for fi in self._field_infos}

self._ref_fields = {fi.name: fi.runtime_ref_tracking for fi in self._field_infos}

self._dynamic_fields = {fi.name: fi.dynamic for fi in self._field_infos}

self._field_metas = {}

elif self._fields_from_typedef:

self._field_names = list(field_names)

self._serializers = list(serializers)

self._nullable_fields = nullable_fields or {}

self._ref_fields = ref_fields or {}

self._dynamic_fields = dynamic_fields or {}

self._field_infos = []

self._field_metas = {}

else:

self._field_infos, self._field_metas = _extract_field_infos(type_resolver, clz, self._type_hints)

if self._field_infos:

self._field_names = [fi.name for fi in self._field_infos]

self._serializers = [fi.serializer for fi in self._field_infos]

self._nullable_fields = {fi.name: fi.nullable for fi in self._field_infos}

self._ref_fields = {fi.name: fi.runtime_ref_tracking for fi in self._field_infos}

self._dynamic_fields = {fi.name: fi.dynamic for fi in self._field_infos}

else:

self._field_names = field_names or self._get_field_names(clz)

self._nullable_fields = nullable_fields or {}

self._ref_fields = {}

self._dynamic_fields = {}

if self._field_names and not self._nullable_fields:

for field_name in self._field_names:

if field_name in self._type_hints:

unwrapped_type, is_optional = unwrap_optional(self._type_hints[field_name])

self._nullable_fields[field_name] = is_optional or not is_primitive_type(unwrapped_type)

self._serializers = serializers or [None] * len(self._field_names)

if serializers is None:

visitor = StructFieldSerializerVisitor(type_resolver)

for index, key in enumerate(self._field_names):

unwrapped_type, _ = unwrap_optional(self._type_hints.get(key, typing.Any))

self._serializers[index] = infer_field(key, unwrapped_type, visitor, types_path=[])

self._unwrapped_hints = self._compute_unwrapped_hints()

if self._fields_from_typedef:

hash_str = compute_struct_fingerprint(self.type_resolver, self._field_names, self._serializers, self._nullable_fields, self._field_infos)

hash_bytes = hash_str.encode("utf-8")

if len(hash_bytes) == 0:

self._hash = 47

else:

full_hash = hash_buffer(hash_bytes, seed=47)[0]

type_hash_32 = full_hash & 0xFFFFFFFF

if full_hash & 0x80000000:

type_hash_32 -= 0x100000000

self._hash = type_hash_32

else:

self._hash, self._field_names, self._serializers = compute_struct_meta(

self.type_resolver, self._field_names, self._serializers, self._nullable_fields, self._field_infos

)

self._field_name_interned = {name: sys.intern(name) for name in self._field_names}

self._current_class_field_names = set(self._get_field_names(self.type_))

self._assign_fields = [

bool(getattr(self._field_infos[index], "assign", True)) if index < len(self._field_infos) else True

for index in range(len(self._field_names))

]

self._validation_field_types = [

getattr(self._field_infos[index], "validation_field_type", None) if index < len(self._field_infos) else None

for index in range(len(self._field_names))

]

self._has_validation_fields = any(field_type is not None for field_type in self._validation_field_types)

self._assigned_field_names = {

field_name

for index, field_name in enumerate(self._field_names)

if self._assign_fields[index] and field_name in self._current_class_field_names

}

self._has_missing_fields = any(

field_name not in self._current_class_field_names or not self._assign_fields[index] for index, field_name in enumerate(self._field_names)

)

self._default_values_factory = (

build_default_values_factory(self.type_resolver, self._type_hints, dataclasses.fields(self.type_))

if dataclasses.is_dataclass(self.type_)

else {}

)

self._missing_field_defaults = self._build_missing_field_defaults()

from pyfory.converter import CompatibleScalarFieldSerializer

self._compatible_scalar_field_flags = [isinstance(serializer, CompatibleScalarFieldSerializer) for serializer in self._serializers]

self._basic_field_flags = [

(not self._dynamic_fields.get(field_name, False))

and not self._compatible_scalar_field_flags[index]

and isinstance(self._serializers[index], self._BASIC_SERIALIZERS)

for index, field_name in enumerate(self._field_names)

]

self.read_data_always_advances = not self.type_resolver.compatible or any(

self._basic_field_flags[index]

or self._nullable_fields.get(field_name, False)

or self._ref_fields.get(field_name, False)

or self._dynamic_fields.get(field_name, False)

or (self._serializers[index] is not None and self._serializers[index].read_data_always_advances)

for index, field_name in enumerate(self._field_names)

)

if self._has_slots:

self._graph_memory_bytes = _SLOTTED_STRUCT_OWNER_BYTES + len(self._field_names) * _REFERENCE_BYTES

else:

# Dict-backed instances retain an instance dict with key and value references per field.

self._graph_memory_bytes = _DICT_BACKED_STRUCT_OWNER_BYTES + _INSTANCE_DICT_OWNER_BYTES + len(self._field_names) * 2 * _REFERENCE_BYTES

def _get_field_names(self, clz):

if hasattr(clz, "__dict__"):

if dataclasses.is_dataclass(clz):

return [field.name for field in dataclasses.fields(clz)]

return sorted(self._type_hints.keys())

if hasattr(clz, "__slots__"):

slots = clz.__slots__

if isinstance(slots, str):

return [slots]

return sorted(slots)

return []

def _compute_unwrapped_hints(self):

return {field_name: unwrap_optional(hint)[0] for field_name, hint in self._type_hints.items()}

def _build_missing_field_defaults(self):

if not self.type_resolver.compatible or not self._default_values_factory:

return []

missing_fields = self._current_class_field_names - self._assigned_field_names

if not missing_fields:

return []

return [(field_name, default_factory) for field_name, default_factory in self._default_values_factory.items() if field_name in missing_fields]

def _write_field_value(self, write_context, serializer, field_value, is_nullable, is_dynamic, is_basic, is_tracking_ref):

if is_basic:

if is_nullable:

if field_value is None:

write_context.write_int8(NULL_FLAG)

else:

write_context.write_int8(NOT_NULL_VALUE_FLAG)

serializer.write(write_context, field_value)

else:

serializer.write(write_context, field_value)

return

if is_tracking_ref:

write_context.write_ref(field_value, serializer=None if is_dynamic else serializer)

return

if is_nullable:

if field_value is None:

write_context.write_int8(NULL_FLAG)

return

write_context.write_int8(NOT_NULL_VALUE_FLAG)

if is_dynamic:

write_context.write_no_ref(field_value)

else:

write_context.write_no_ref(field_value, serializer=serializer)

def _read_field_value(

self,

read_context,

serializer,

is_nullable,

is_dynamic,

is_basic,

is_tracking_ref,

is_compatible_scalar_field,

):

if is_nullable and is_basic:

if read_context.read_int8() == NULL_FLAG:

return None

return serializer.read(read_context)

if is_basic:

return serializer.read(read_context)

if is_tracking_ref:

return read_context.read_ref(serializer=None if is_dynamic else serializer)

if is_nullable:

flag = read_context.read_int8()

if flag == NULL_FLAG:

return None

if is_compatible_scalar_field and flag != NOT_NULL_VALUE_FLAG:

from pyfory.error import ForyInvalidDataError

raise ForyInvalidDataError(f"Invalid compatible scalar null flag: {flag}")

if is_dynamic:

return read_context.read_no_ref()

return read_context.read_no_ref(serializer=serializer)

def _default_field_value(self, field_name):

default_factory = self._default_values_factory.get(field_name)

if default_factory is None:

return None

return default_factory()

def _assign_read_field_value(self, obj, obj_dict, field_name, field_value, validation_field_type):

if validation_field_type is not None:

from pyfory.meta.typedef import coerce_assignable_value, is_value_assignable

if not is_value_assignable(field_value, validation_field_type):

field_value = self._default_field_value(field_name)

else:

field_value = coerce_assignable_value(field_value, validation_field_type)

interned_name = self._field_name_interned[field_name]

if obj_dict is not None:

obj_dict[interned_name] = field_value

else:

setattr(obj, interned_name, field_value)

def write(self, write_context: Buffer, value):

if not self.type_resolver.compatible:

write_context.write_int32(self._hash)

value_dict = value.__dict__ if not self._has_slots else None

if value_dict is not None:

if self.type_resolver.compatible:

for index, field_name in enumerate(self._field_names):

interned_name = self._field_name_interned[field_name]

field_value = value_dict.get(interned_name)

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

self._write_field_value(write_context, serializer, field_value, is_nullable, is_dynamic, is_basic, is_tracking_ref)

else:

for index, field_name in enumerate(self._field_names):

interned_name = self._field_name_interned[field_name]

field_value = value_dict[interned_name]

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

self._write_field_value(write_context, serializer, field_value, is_nullable, is_dynamic, is_basic, is_tracking_ref)

else:

if self.type_resolver.compatible:

for index, field_name in enumerate(self._field_names):

interned_name = self._field_name_interned[field_name]

field_value = getattr(value, interned_name, None)

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

self._write_field_value(write_context, serializer, field_value, is_nullable, is_dynamic, is_basic, is_tracking_ref)

else:

for index, field_name in enumerate(self._field_names):

interned_name = self._field_name_interned[field_name]

field_value = getattr(value, interned_name)

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

self._write_field_value(write_context, serializer, field_value, is_nullable, is_dynamic, is_basic, is_tracking_ref)

write_context.try_flush()

def read(self, read_context):

if read_context.policy is not DEFAULT_POLICY:

read_context.policy.authorize_instantiation(self.type_)

if not self.type_resolver.compatible:

hash_ = read_context.read_int32()

if hash_ != self._hash:

raise TypeNotCompatibleError(

f"Hash {hash_} is not consistent with {self._hash} for type {self.type_}",

)

read_context.reserve_graph_memory(self._graph_memory_bytes)

obj = self.type_.__new__(self.type_)

read_context.reference(obj)

obj_dict = obj.__dict__ if not self._has_slots else None

if self._has_missing_fields:

for index, field_name in enumerate(self._field_names):

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

is_compatible_scalar_field = self._compatible_scalar_field_flags[index]

if field_name not in self._current_class_field_names or not self._assign_fields[index]:

self._read_field_value(

read_context,

serializer,

is_nullable,

is_dynamic,

is_basic,

is_tracking_ref,

is_compatible_scalar_field,

)

continue

field_value = self._read_field_value(

read_context,

serializer,

is_nullable,

is_dynamic,

is_basic,

is_tracking_ref,

is_compatible_scalar_field,

)

validation_field_type = self._validation_field_types[index]

if validation_field_type is None:

interned_name = self._field_name_interned[field_name]

if obj_dict is not None:

obj_dict[interned_name] = field_value

else:

setattr(obj, interned_name, field_value)

else:

self._assign_read_field_value(obj, obj_dict, field_name, field_value, validation_field_type)

else:

if not self._has_validation_fields:

for index, field_name in enumerate(self._field_names):

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

is_compatible_scalar_field = self._compatible_scalar_field_flags[index]

field_value = self._read_field_value(

read_context,

serializer,

is_nullable,

is_dynamic,

is_basic,

is_tracking_ref,

is_compatible_scalar_field,

)

interned_name = self._field_name_interned[field_name]

if obj_dict is not None:

obj_dict[interned_name] = field_value

else:

setattr(obj, interned_name, field_value)

else:

for index, field_name in enumerate(self._field_names):

serializer = self._serializers[index]

is_nullable = self._nullable_fields.get(field_name, False)

is_dynamic = self._dynamic_fields.get(field_name, False)

is_tracking_ref = self._ref_fields.get(field_name, False)

is_basic = self._basic_field_flags[index]

is_compatible_scalar_field = self._compatible_scalar_field_flags[index]

field_value = self._read_field_value(

read_context,

serializer,

is_nullable,

is_dynamic,

is_basic,

is_tracking_ref,

is_compatible_scalar_field,

)

validation_field_type = self._validation_field_types[index]

if validation_field_type is None:

interned_name = self._field_name_interned[field_name]

if obj_dict is not None:

obj_dict[interned_name] = field_value

else:

setattr(obj, interned_name, field_value)

else:

self._assign_read_field_value(obj, obj_dict, field_name, field_value, validation_field_type)

if self._missing_field_defaults:

for field_name, default_factory in self._missing_field_defaults:

value = default_factory()

if obj_dict is not None:

obj_dict[field_name] = value

else:

setattr(obj, field_name, value)

return obj

class DataClassStubSerializer(DataClassSerializer):

def __init__(self, type_resolver, clz: type):

Serializer.__init__(self, type_resolver, clz)

def write(self, write_context, value):

self._replace().write(write_context, value)

def read(self, read_context):

return self._replace().read(read_context)

def _replace(self):

typeinfo = self.type_resolver.get_type_info(self.type_)

typeinfo.serializer = DataClassSerializer(self.type_resolver, self.type_)

return typeinfo.serializer

basic_types = {

bool,

int,

float,

str,

bytes,

datetime.datetime,

datetime.date,

datetime.time,

datetime.timedelta,

decimal.Decimal,

TypeId.INT8,

TypeId.INT16,

TypeId.VARINT32,

TypeId.INT32,

TypeId.VARINT64,

TypeId.INT64,

TypeId.TAGGED_INT64,

TypeId.UINT8,

TypeId.UINT16,

TypeId.VAR_UINT32,