in serde-generate/runtime/python/serde_binary/__init__.py [0:0]
def deserialize_any(self, obj_type) -> typing.Any:
if obj_type in self.primitive_type_deserializer:
return self.primitive_type_deserializer[obj_type]()
elif hasattr(obj_type, "__origin__"): # Generic type
types = getattr(obj_type, "__args__")
if getattr(obj_type, "__origin__") == collections.abc.Sequence: # Sequence
assert len(types) == 1
item_type = types[0]
length = self.deserialize_len()
result = []
for i in range(0, length):
item = self.deserialize_any(item_type)
result.append(item)
return result
elif getattr(obj_type, "__origin__") == tuple: # Tuple
result = []
if len(types) is 1 and types[0] is ():
return tuple()
for i in range(len(types)):
item = self.deserialize_any(types[i])
result.append(item)
return tuple(result)
elif getattr(obj_type, "__origin__") == typing.Union: # Option
assert len(types) == 2 and types[1] == type(None)
tag = int.from_bytes(self.read(1), byteorder="little", signed=False)
if tag == 0:
return None
elif tag == 1:
return self.deserialize_any(types[0])
else:
raise st.DeserializationError("Wrong tag for Option value")
elif getattr(obj_type, "__origin__") == dict: # Map
assert len(types) == 2
length = self.deserialize_len()
result = dict()
previous_key_slice = None
for i in range(0, length):
key_start = self.get_buffer_offset()
key = self.deserialize_any(types[0])
key_end = self.get_buffer_offset()
value = self.deserialize_any(types[1])
key_slice = (key_start, key_end)
if previous_key_slice is not None:
self.check_that_key_slices_are_increasing(
previous_key_slice, key_slice
)
previous_key_slice = key_slice
result[key] = value
return result
else:
raise st.DeserializationError("Unexpected type", obj_type)
else:
# handle structs
if dataclasses.is_dataclass(obj_type):
values = []
fields = dataclasses.fields(obj_type)
typing_hints = get_type_hints(obj_type)
self.increase_container_depth()
for field in fields:
field_type = typing_hints[field.name]
field_value = self.deserialize_any(field_type)
values.append(field_value)
self.decrease_container_depth()
return obj_type(*values)
# handle variant
elif hasattr(obj_type, "VARIANTS"):
variant_index = self.deserialize_variant_index()
if variant_index not in range(len(obj_type.VARIANTS)):
raise st.DeserializationError(
"Unexpected variant index", variant_index
)
new_type = obj_type.VARIANTS[variant_index]
return self.deserialize_any(new_type)
else:
raise st.DeserializationError("Unexpected type", obj_type)