Module libra.serde_binary

Module describing the "binary" serialization formats.

Note: This internal module is currently only meant to share code between the LCS and bincode formats. Internal APIs could change in the future.

Expand source code
# Copyright (c) Facebook, Inc. and its affiliates
# SPDX-License-Identifier: MIT OR Apache-2.0

"""
Module describing the "binary" serialization formats.

Note: This internal module is currently only meant to share code between the LCS and bincode formats. Internal APIs could change in the future.
"""

import dataclasses
import collections
import io
import typing
from typing import get_type_hints

from libra import serde_types as st


@dataclasses.dataclass
class BinarySerializer:
    """Serialization primitives for binary formats (abstract class).

    "Binary" serialization formats may differ in the way they encode sequence lengths, variant
    index, and how they sort map entries (or not).
    """

    output: io.BytesIO
    container_depth_budget: typing.Optional[int]

    def serialize_bytes(self, value: bytes):
        self.serialize_len(len(value))
        self.output.write(value)

    def serialize_str(self, value: str):
        self.serialize_bytes(value.encode())

    def serialize_unit(self, value: st.unit):
        pass

    def serialize_bool(self, value: st.bool):
        self.output.write(int(value).to_bytes(1, "little", signed=False))

    def serialize_u8(self, value: st.uint8):
        self.output.write(int(value).to_bytes(1, "little", signed=False))

    def serialize_u16(self, value: st.uint16):
        self.output.write(int(value).to_bytes(2, "little", signed=False))

    def serialize_u32(self, value: st.uint32):
        self.output.write(int(value).to_bytes(4, "little", signed=False))

    def serialize_u64(self, value: st.uint64):
        self.output.write(int(value).to_bytes(8, "little", signed=False))

    def serialize_u128(self, value: st.uint128):
        self.output.write(int(value).to_bytes(16, "little", signed=False))

    def serialize_i8(self, value: st.uint8):
        self.output.write(int(value).to_bytes(1, "little", signed=True))

    def serialize_i16(self, value: st.uint16):
        self.output.write(int(value).to_bytes(2, "little", signed=True))

    def serialize_i32(self, value: st.uint32):
        self.output.write(int(value).to_bytes(4, "little", signed=True))

    def serialize_i64(self, value: st.uint64):
        self.output.write(int(value).to_bytes(8, "little", signed=True))

    def serialize_i128(self, value: st.uint128):
        self.output.write(int(value).to_bytes(16, "little", signed=True))

    def serialize_f32(self, value: st.float32):
        raise NotImplementedError

    def serialize_f64(self, value: st.float64):
        raise NotImplementedError

    def serialize_char(self, value: st.char):
        raise NotImplementedError

    def get_buffer_offset(self) -> int:
        return len(self.output.getbuffer())

    def get_buffer(self) -> bytes:
        return self.output.getvalue()

    def increase_container_depth(self):
        if self.container_depth_budget is not None:
            if self.container_depth_budget == 0:
                raise st.SerializationError("Exceeded maximum container depth")
            self.container_depth_budget -= 1

    def decrease_container_depth(self):
        if self.container_depth_budget is not None:
            self.container_depth_budget += 1

    def serialize_len(self, value: int):
        raise NotImplementedError

    def serialize_variant_index(self, value: int):
        raise NotImplementedError

    def sort_map_entries(self, offsets: typing.List[int]):
        raise NotImplementedError

    PRIMITIVE_TYPE_SERIALIZER = {
        st.bool: serialize_bool,
        st.uint8: serialize_u8,
        st.uint16: serialize_u16,
        st.uint32: serialize_u32,
        st.uint64: serialize_u64,
        st.uint128: serialize_u128,
        st.int8: serialize_i8,
        st.int16: serialize_i16,
        st.int32: serialize_i32,
        st.int64: serialize_i64,
        st.int128: serialize_i128,
        st.float32: serialize_f32,
        st.float64: serialize_f64,
        st.unit: serialize_unit,
        st.char: serialize_char,
        str: serialize_str,
        bytes: serialize_bytes,
    }

    # noqa: C901
    def serialize_any(self, obj: typing.Any, obj_type):
        if obj_type in self.PRIMITIVE_TYPE_SERIALIZER:
            self.PRIMITIVE_TYPE_SERIALIZER[obj_type](self, obj)

        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]
                self.serialize_len(len(obj))
                for item in obj:
                    self.serialize_any(item, item_type)

            elif getattr(obj_type, "__origin__") == tuple:  # Tuple
                for i in range(len(obj)):
                    self.serialize_any(obj[i], types[i])

            elif getattr(obj_type, "__origin__") == typing.Union:  # Option
                assert len(types) == 2 and types[1] == type(None)
                if obj is None:
                    self.output.write(b"\x00")
                else:
                    self.output.write(b"\x01")
                    self.serialize_any(obj, types[0])

            elif getattr(obj_type, "__origin__") == dict:  # Map
                assert len(types) == 2
                self.serialize_len(len(obj))
                offsets = []
                for key, value in obj.items():
                    offsets.append(self.get_buffer_offset())
                    self.serialize_any(key, types[0])
                    self.serialize_any(value, types[1])
                self.sort_map_entries(offsets)

            else:
                raise st.SerializationError("Unexpected type", obj_type)

        else:
            if not dataclasses.is_dataclass(obj_type):  # Enum
                if not hasattr(obj_type, "VARIANTS"):
                    raise st.SerializationError("Unexpected type", obj_type)
                if not hasattr(obj, "INDEX"):
                    raise st.SerializationError("Wrong Value for the type", obj, obj_type)
                self.serialize_variant_index(obj.__class__.INDEX)
                # Proceed to variant
                obj_type = obj_type.VARIANTS[obj.__class__.INDEX]
                if not dataclasses.is_dataclass(obj_type):
                    raise st.SerializationError("Unexpected type", obj_type)

            # pyre-ignore
            if not isinstance(obj, obj_type):
                raise st.SerializationError("Wrong Value for the type", obj, obj_type)

            # Content of struct or variant
            fields = dataclasses.fields(obj_type)
            types = get_type_hints(obj_type)
            self.increase_container_depth()
            for field in fields:
                field_value = obj.__dict__[field.name]
                field_type = types[field.name]
                self.serialize_any(field_value, field_type)
            self.decrease_container_depth()


@dataclasses.dataclass
class BinaryDeserializer:
    """Deserialization primitives for binary formats (abstract class).

    "Binary" serialization formats may differ in the way they encode sequence lengths, variant
    index, and how they verify the ordering of keys in map entries (or not).
    """

    input: io.BytesIO
    container_depth_budget: typing.Optional[int]

    def read(self, length: int) -> bytes:
        value = self.input.read(length)
        if value is None or len(value) < length:
            raise st.DeserializationError("Input is too short")
        return value

    def deserialize_bytes(self) -> bytes:
        length = self.deserialize_len()
        return self.read(length)

    def deserialize_str(self) -> str:
        content = self.deserialize_bytes()
        try:
            return content.decode()
        except UnicodeDecodeError:
            raise st.DeserializationError("Invalid unicode string:", content)

    def deserialize_unit(self) -> st.unit:
        pass

    def deserialize_bool(self) -> st.bool:
        b = int.from_bytes(self.read(1), byteorder="little", signed=False)
        if b == 0:
            return False
        elif b == 1:
            return True
        else:
            raise st.DeserializationError("Unexpected boolean value:", b)

    def deserialize_u8(self) -> st.uint8:
        return st.uint8(int.from_bytes(self.read(1), byteorder="little", signed=False))

    def deserialize_u16(self) -> st.uint16:
        return st.uint16(int.from_bytes(self.read(2), byteorder="little", signed=False))

    def deserialize_u32(self) -> st.uint32:
        return st.uint32(int.from_bytes(self.read(4), byteorder="little", signed=False))

    def deserialize_u64(self) -> st.uint64:
        return st.uint64(int.from_bytes(self.read(8), byteorder="little", signed=False))

    def deserialize_u128(self) -> st.uint128:
        return st.uint128(int.from_bytes(self.read(16), byteorder="little", signed=False))

    def deserialize_i8(self) -> st.int8:
        return st.int8(int.from_bytes(self.read(1), byteorder="little", signed=True))

    def deserialize_i16(self) -> st.int16:
        return st.int16(int.from_bytes(self.read(2), byteorder="little", signed=True))

    def deserialize_i32(self) -> st.int32:
        return st.int32(int.from_bytes(self.read(4), byteorder="little", signed=True))

    def deserialize_i64(self) -> st.int64:
        return st.int64(int.from_bytes(self.read(8), byteorder="little", signed=True))

    def deserialize_i128(self) -> st.int128:
        return st.int128(int.from_bytes(self.read(16), byteorder="little", signed=True))

    def deserialize_f32(self) -> st.float32:
        raise NotImplementedError

    def deserialize_f64(self) -> st.float64:
        raise NotImplementedError

    def deserialize_char(self) -> st.char:
        raise NotImplementedError

    def get_buffer_offset(self) -> int:
        return self.input.tell()

    def get_remaining_buffer(self) -> bytes:
        buf = self.input.getbuffer()
        offset = self.get_buffer_offset()
        return bytes(buf[offset:])

    def increase_container_depth(self):
        if self.container_depth_budget is not None:
            if self.container_depth_budget == 0:
                raise st.DeserializationError("Exceeded maximum container depth")
            self.container_depth_budget -= 1

    def decrease_container_depth(self):
        if self.container_depth_budget is not None:
            self.container_depth_budget += 1

    def deserialize_len(self) -> int:
        raise NotImplementedError

    def deserialize_variant_index(self) -> int:
        raise NotImplementedError

    def check_that_key_slices_are_increasing(
        self, slice1: typing.Tuple[int, int], slice2: typing.Tuple[int, int]
    ) -> bool:
        raise NotImplementedError

    PRIMITIVE_TYPE_DESERIALIZER = {
        st.bool: deserialize_bool,
        st.uint8: deserialize_u8,
        st.uint16: deserialize_u16,
        st.uint32: deserialize_u32,
        st.uint64: deserialize_u64,
        st.uint128: deserialize_u128,
        st.int8: deserialize_i8,
        st.int16: deserialize_i16,
        st.int32: deserialize_i32,
        st.int64: deserialize_i64,
        st.int128: deserialize_i128,
        st.float32: deserialize_f32,
        st.float64: deserialize_f64,
        st.unit: deserialize_unit,
        st.char: deserialize_char,
        str: deserialize_str,
        bytes: deserialize_bytes,
    }

    # noqa
    def deserialize_any(self, obj_type) -> typing.Any:
        if obj_type in BinaryDeserializer.PRIMITIVE_TYPE_DESERIALIZER:
            return BinaryDeserializer.PRIMITIVE_TYPE_DESERIALIZER[obj_type](self)

        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 = []
                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)

Classes

class BinaryDeserializer (input: _io.BytesIO, container_depth_budget: Union[int, NoneType])

Deserialization primitives for binary formats (abstract class).

"Binary" serialization formats may differ in the way they encode sequence lengths, variant index, and how they verify the ordering of keys in map entries (or not).

Expand source code
class BinaryDeserializer:
    """Deserialization primitives for binary formats (abstract class).

    "Binary" serialization formats may differ in the way they encode sequence lengths, variant
    index, and how they verify the ordering of keys in map entries (or not).
    """

    input: io.BytesIO
    container_depth_budget: typing.Optional[int]

    def read(self, length: int) -> bytes:
        value = self.input.read(length)
        if value is None or len(value) < length:
            raise st.DeserializationError("Input is too short")
        return value

    def deserialize_bytes(self) -> bytes:
        length = self.deserialize_len()
        return self.read(length)

    def deserialize_str(self) -> str:
        content = self.deserialize_bytes()
        try:
            return content.decode()
        except UnicodeDecodeError:
            raise st.DeserializationError("Invalid unicode string:", content)

    def deserialize_unit(self) -> st.unit:
        pass

    def deserialize_bool(self) -> st.bool:
        b = int.from_bytes(self.read(1), byteorder="little", signed=False)
        if b == 0:
            return False
        elif b == 1:
            return True
        else:
            raise st.DeserializationError("Unexpected boolean value:", b)

    def deserialize_u8(self) -> st.uint8:
        return st.uint8(int.from_bytes(self.read(1), byteorder="little", signed=False))

    def deserialize_u16(self) -> st.uint16:
        return st.uint16(int.from_bytes(self.read(2), byteorder="little", signed=False))

    def deserialize_u32(self) -> st.uint32:
        return st.uint32(int.from_bytes(self.read(4), byteorder="little", signed=False))

    def deserialize_u64(self) -> st.uint64:
        return st.uint64(int.from_bytes(self.read(8), byteorder="little", signed=False))

    def deserialize_u128(self) -> st.uint128:
        return st.uint128(int.from_bytes(self.read(16), byteorder="little", signed=False))

    def deserialize_i8(self) -> st.int8:
        return st.int8(int.from_bytes(self.read(1), byteorder="little", signed=True))

    def deserialize_i16(self) -> st.int16:
        return st.int16(int.from_bytes(self.read(2), byteorder="little", signed=True))

    def deserialize_i32(self) -> st.int32:
        return st.int32(int.from_bytes(self.read(4), byteorder="little", signed=True))

    def deserialize_i64(self) -> st.int64:
        return st.int64(int.from_bytes(self.read(8), byteorder="little", signed=True))

    def deserialize_i128(self) -> st.int128:
        return st.int128(int.from_bytes(self.read(16), byteorder="little", signed=True))

    def deserialize_f32(self) -> st.float32:
        raise NotImplementedError

    def deserialize_f64(self) -> st.float64:
        raise NotImplementedError

    def deserialize_char(self) -> st.char:
        raise NotImplementedError

    def get_buffer_offset(self) -> int:
        return self.input.tell()

    def get_remaining_buffer(self) -> bytes:
        buf = self.input.getbuffer()
        offset = self.get_buffer_offset()
        return bytes(buf[offset:])

    def increase_container_depth(self):
        if self.container_depth_budget is not None:
            if self.container_depth_budget == 0:
                raise st.DeserializationError("Exceeded maximum container depth")
            self.container_depth_budget -= 1

    def decrease_container_depth(self):
        if self.container_depth_budget is not None:
            self.container_depth_budget += 1

    def deserialize_len(self) -> int:
        raise NotImplementedError

    def deserialize_variant_index(self) -> int:
        raise NotImplementedError

    def check_that_key_slices_are_increasing(
        self, slice1: typing.Tuple[int, int], slice2: typing.Tuple[int, int]
    ) -> bool:
        raise NotImplementedError

    PRIMITIVE_TYPE_DESERIALIZER = {
        st.bool: deserialize_bool,
        st.uint8: deserialize_u8,
        st.uint16: deserialize_u16,
        st.uint32: deserialize_u32,
        st.uint64: deserialize_u64,
        st.uint128: deserialize_u128,
        st.int8: deserialize_i8,
        st.int16: deserialize_i16,
        st.int32: deserialize_i32,
        st.int64: deserialize_i64,
        st.int128: deserialize_i128,
        st.float32: deserialize_f32,
        st.float64: deserialize_f64,
        st.unit: deserialize_unit,
        st.char: deserialize_char,
        str: deserialize_str,
        bytes: deserialize_bytes,
    }

    # noqa
    def deserialize_any(self, obj_type) -> typing.Any:
        if obj_type in BinaryDeserializer.PRIMITIVE_TYPE_DESERIALIZER:
            return BinaryDeserializer.PRIMITIVE_TYPE_DESERIALIZER[obj_type](self)

        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 = []
                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)

Subclasses

Class variables

var PRIMITIVE_TYPE_DESERIALIZER
var container_depth_budget : Union[int, NoneType]
var input : _io.BytesIO

Methods

def check_that_key_slices_are_increasing(self, slice1: Tuple[int, int], slice2: Tuple[int, int]) ‑> bool
Expand source code
def check_that_key_slices_are_increasing(
    self, slice1: typing.Tuple[int, int], slice2: typing.Tuple[int, int]
) -> bool:
    raise NotImplementedError
def decrease_container_depth(self)
Expand source code
def decrease_container_depth(self):
    if self.container_depth_budget is not None:
        self.container_depth_budget += 1
def deserialize_any(self, obj_type) ‑> Any
Expand source code
def deserialize_any(self, obj_type) -> typing.Any:
    if obj_type in BinaryDeserializer.PRIMITIVE_TYPE_DESERIALIZER:
        return BinaryDeserializer.PRIMITIVE_TYPE_DESERIALIZER[obj_type](self)

    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 = []
            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)
def deserialize_bool(self) ‑> bool
Expand source code
def deserialize_bool(self) -> st.bool:
    b = int.from_bytes(self.read(1), byteorder="little", signed=False)
    if b == 0:
        return False
    elif b == 1:
        return True
    else:
        raise st.DeserializationError("Unexpected boolean value:", b)
def deserialize_bytes(self) ‑> bytes
Expand source code
def deserialize_bytes(self) -> bytes:
    length = self.deserialize_len()
    return self.read(length)
def deserialize_char(self) ‑> char
Expand source code
def deserialize_char(self) -> st.char:
    raise NotImplementedError
def deserialize_f32(self) ‑> numpy.float32
Expand source code
def deserialize_f32(self) -> st.float32:
    raise NotImplementedError
def deserialize_f64(self) ‑> numpy.float64
Expand source code
def deserialize_f64(self) -> st.float64:
    raise NotImplementedError
def deserialize_i128(self) ‑> int128
Expand source code
def deserialize_i128(self) -> st.int128:
    return st.int128(int.from_bytes(self.read(16), byteorder="little", signed=True))
def deserialize_i16(self) ‑> numpy.int16
Expand source code
def deserialize_i16(self) -> st.int16:
    return st.int16(int.from_bytes(self.read(2), byteorder="little", signed=True))
def deserialize_i32(self) ‑> numpy.int32
Expand source code
def deserialize_i32(self) -> st.int32:
    return st.int32(int.from_bytes(self.read(4), byteorder="little", signed=True))
def deserialize_i64(self) ‑> numpy.int64
Expand source code
def deserialize_i64(self) -> st.int64:
    return st.int64(int.from_bytes(self.read(8), byteorder="little", signed=True))
def deserialize_i8(self) ‑> numpy.int8
Expand source code
def deserialize_i8(self) -> st.int8:
    return st.int8(int.from_bytes(self.read(1), byteorder="little", signed=True))
def deserialize_len(self) ‑> int
Expand source code
def deserialize_len(self) -> int:
    raise NotImplementedError
def deserialize_str(self) ‑> str
Expand source code
def deserialize_str(self) -> str:
    content = self.deserialize_bytes()
    try:
        return content.decode()
    except UnicodeDecodeError:
        raise st.DeserializationError("Invalid unicode string:", content)
def deserialize_u128(self) ‑> uint128
Expand source code
def deserialize_u128(self) -> st.uint128:
    return st.uint128(int.from_bytes(self.read(16), byteorder="little", signed=False))
def deserialize_u16(self) ‑> numpy.uint16
Expand source code
def deserialize_u16(self) -> st.uint16:
    return st.uint16(int.from_bytes(self.read(2), byteorder="little", signed=False))
def deserialize_u32(self) ‑> numpy.uint32
Expand source code
def deserialize_u32(self) -> st.uint32:
    return st.uint32(int.from_bytes(self.read(4), byteorder="little", signed=False))
def deserialize_u64(self) ‑> numpy.uint64
Expand source code
def deserialize_u64(self) -> st.uint64:
    return st.uint64(int.from_bytes(self.read(8), byteorder="little", signed=False))
def deserialize_u8(self) ‑> numpy.uint8
Expand source code
def deserialize_u8(self) -> st.uint8:
    return st.uint8(int.from_bytes(self.read(1), byteorder="little", signed=False))
def deserialize_unit(self) ‑> Type[NoneType]
Expand source code
def deserialize_unit(self) -> st.unit:
    pass
def deserialize_variant_index(self) ‑> int
Expand source code
def deserialize_variant_index(self) -> int:
    raise NotImplementedError
def get_buffer_offset(self) ‑> int
Expand source code
def get_buffer_offset(self) -> int:
    return self.input.tell()
def get_remaining_buffer(self) ‑> bytes
Expand source code
def get_remaining_buffer(self) -> bytes:
    buf = self.input.getbuffer()
    offset = self.get_buffer_offset()
    return bytes(buf[offset:])
def increase_container_depth(self)
Expand source code
def increase_container_depth(self):
    if self.container_depth_budget is not None:
        if self.container_depth_budget == 0:
            raise st.DeserializationError("Exceeded maximum container depth")
        self.container_depth_budget -= 1
def read(self, length: int) ‑> bytes
Expand source code
def read(self, length: int) -> bytes:
    value = self.input.read(length)
    if value is None or len(value) < length:
        raise st.DeserializationError("Input is too short")
    return value
class BinarySerializer (output: _io.BytesIO, container_depth_budget: Union[int, NoneType])

Serialization primitives for binary formats (abstract class).

"Binary" serialization formats may differ in the way they encode sequence lengths, variant index, and how they sort map entries (or not).

Expand source code
class BinarySerializer:
    """Serialization primitives for binary formats (abstract class).

    "Binary" serialization formats may differ in the way they encode sequence lengths, variant
    index, and how they sort map entries (or not).
    """

    output: io.BytesIO
    container_depth_budget: typing.Optional[int]

    def serialize_bytes(self, value: bytes):
        self.serialize_len(len(value))
        self.output.write(value)

    def serialize_str(self, value: str):
        self.serialize_bytes(value.encode())

    def serialize_unit(self, value: st.unit):
        pass

    def serialize_bool(self, value: st.bool):
        self.output.write(int(value).to_bytes(1, "little", signed=False))

    def serialize_u8(self, value: st.uint8):
        self.output.write(int(value).to_bytes(1, "little", signed=False))

    def serialize_u16(self, value: st.uint16):
        self.output.write(int(value).to_bytes(2, "little", signed=False))

    def serialize_u32(self, value: st.uint32):
        self.output.write(int(value).to_bytes(4, "little", signed=False))

    def serialize_u64(self, value: st.uint64):
        self.output.write(int(value).to_bytes(8, "little", signed=False))

    def serialize_u128(self, value: st.uint128):
        self.output.write(int(value).to_bytes(16, "little", signed=False))

    def serialize_i8(self, value: st.uint8):
        self.output.write(int(value).to_bytes(1, "little", signed=True))

    def serialize_i16(self, value: st.uint16):
        self.output.write(int(value).to_bytes(2, "little", signed=True))

    def serialize_i32(self, value: st.uint32):
        self.output.write(int(value).to_bytes(4, "little", signed=True))

    def serialize_i64(self, value: st.uint64):
        self.output.write(int(value).to_bytes(8, "little", signed=True))

    def serialize_i128(self, value: st.uint128):
        self.output.write(int(value).to_bytes(16, "little", signed=True))

    def serialize_f32(self, value: st.float32):
        raise NotImplementedError

    def serialize_f64(self, value: st.float64):
        raise NotImplementedError

    def serialize_char(self, value: st.char):
        raise NotImplementedError

    def get_buffer_offset(self) -> int:
        return len(self.output.getbuffer())

    def get_buffer(self) -> bytes:
        return self.output.getvalue()

    def increase_container_depth(self):
        if self.container_depth_budget is not None:
            if self.container_depth_budget == 0:
                raise st.SerializationError("Exceeded maximum container depth")
            self.container_depth_budget -= 1

    def decrease_container_depth(self):
        if self.container_depth_budget is not None:
            self.container_depth_budget += 1

    def serialize_len(self, value: int):
        raise NotImplementedError

    def serialize_variant_index(self, value: int):
        raise NotImplementedError

    def sort_map_entries(self, offsets: typing.List[int]):
        raise NotImplementedError

    PRIMITIVE_TYPE_SERIALIZER = {
        st.bool: serialize_bool,
        st.uint8: serialize_u8,
        st.uint16: serialize_u16,
        st.uint32: serialize_u32,
        st.uint64: serialize_u64,
        st.uint128: serialize_u128,
        st.int8: serialize_i8,
        st.int16: serialize_i16,
        st.int32: serialize_i32,
        st.int64: serialize_i64,
        st.int128: serialize_i128,
        st.float32: serialize_f32,
        st.float64: serialize_f64,
        st.unit: serialize_unit,
        st.char: serialize_char,
        str: serialize_str,
        bytes: serialize_bytes,
    }

    # noqa: C901
    def serialize_any(self, obj: typing.Any, obj_type):
        if obj_type in self.PRIMITIVE_TYPE_SERIALIZER:
            self.PRIMITIVE_TYPE_SERIALIZER[obj_type](self, obj)

        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]
                self.serialize_len(len(obj))
                for item in obj:
                    self.serialize_any(item, item_type)

            elif getattr(obj_type, "__origin__") == tuple:  # Tuple
                for i in range(len(obj)):
                    self.serialize_any(obj[i], types[i])

            elif getattr(obj_type, "__origin__") == typing.Union:  # Option
                assert len(types) == 2 and types[1] == type(None)
                if obj is None:
                    self.output.write(b"\x00")
                else:
                    self.output.write(b"\x01")
                    self.serialize_any(obj, types[0])

            elif getattr(obj_type, "__origin__") == dict:  # Map
                assert len(types) == 2
                self.serialize_len(len(obj))
                offsets = []
                for key, value in obj.items():
                    offsets.append(self.get_buffer_offset())
                    self.serialize_any(key, types[0])
                    self.serialize_any(value, types[1])
                self.sort_map_entries(offsets)

            else:
                raise st.SerializationError("Unexpected type", obj_type)

        else:
            if not dataclasses.is_dataclass(obj_type):  # Enum
                if not hasattr(obj_type, "VARIANTS"):
                    raise st.SerializationError("Unexpected type", obj_type)
                if not hasattr(obj, "INDEX"):
                    raise st.SerializationError("Wrong Value for the type", obj, obj_type)
                self.serialize_variant_index(obj.__class__.INDEX)
                # Proceed to variant
                obj_type = obj_type.VARIANTS[obj.__class__.INDEX]
                if not dataclasses.is_dataclass(obj_type):
                    raise st.SerializationError("Unexpected type", obj_type)

            # pyre-ignore
            if not isinstance(obj, obj_type):
                raise st.SerializationError("Wrong Value for the type", obj, obj_type)

            # Content of struct or variant
            fields = dataclasses.fields(obj_type)
            types = get_type_hints(obj_type)
            self.increase_container_depth()
            for field in fields:
                field_value = obj.__dict__[field.name]
                field_type = types[field.name]
                self.serialize_any(field_value, field_type)
            self.decrease_container_depth()

Subclasses

Class variables

var PRIMITIVE_TYPE_SERIALIZER
var container_depth_budget : Union[int, NoneType]
var output : _io.BytesIO

Methods

def decrease_container_depth(self)
Expand source code
def decrease_container_depth(self):
    if self.container_depth_budget is not None:
        self.container_depth_budget += 1
def get_buffer(self) ‑> bytes
Expand source code
def get_buffer(self) -> bytes:
    return self.output.getvalue()
def get_buffer_offset(self) ‑> int
Expand source code
def get_buffer_offset(self) -> int:
    return len(self.output.getbuffer())
def increase_container_depth(self)
Expand source code
def increase_container_depth(self):
    if self.container_depth_budget is not None:
        if self.container_depth_budget == 0:
            raise st.SerializationError("Exceeded maximum container depth")
        self.container_depth_budget -= 1
def serialize_any(self, obj: Any, obj_type)
Expand source code
def serialize_any(self, obj: typing.Any, obj_type):
    if obj_type in self.PRIMITIVE_TYPE_SERIALIZER:
        self.PRIMITIVE_TYPE_SERIALIZER[obj_type](self, obj)

    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]
            self.serialize_len(len(obj))
            for item in obj:
                self.serialize_any(item, item_type)

        elif getattr(obj_type, "__origin__") == tuple:  # Tuple
            for i in range(len(obj)):
                self.serialize_any(obj[i], types[i])

        elif getattr(obj_type, "__origin__") == typing.Union:  # Option
            assert len(types) == 2 and types[1] == type(None)
            if obj is None:
                self.output.write(b"\x00")
            else:
                self.output.write(b"\x01")
                self.serialize_any(obj, types[0])

        elif getattr(obj_type, "__origin__") == dict:  # Map
            assert len(types) == 2
            self.serialize_len(len(obj))
            offsets = []
            for key, value in obj.items():
                offsets.append(self.get_buffer_offset())
                self.serialize_any(key, types[0])
                self.serialize_any(value, types[1])
            self.sort_map_entries(offsets)

        else:
            raise st.SerializationError("Unexpected type", obj_type)

    else:
        if not dataclasses.is_dataclass(obj_type):  # Enum
            if not hasattr(obj_type, "VARIANTS"):
                raise st.SerializationError("Unexpected type", obj_type)
            if not hasattr(obj, "INDEX"):
                raise st.SerializationError("Wrong Value for the type", obj, obj_type)
            self.serialize_variant_index(obj.__class__.INDEX)
            # Proceed to variant
            obj_type = obj_type.VARIANTS[obj.__class__.INDEX]
            if not dataclasses.is_dataclass(obj_type):
                raise st.SerializationError("Unexpected type", obj_type)

        # pyre-ignore
        if not isinstance(obj, obj_type):
            raise st.SerializationError("Wrong Value for the type", obj, obj_type)

        # Content of struct or variant
        fields = dataclasses.fields(obj_type)
        types = get_type_hints(obj_type)
        self.increase_container_depth()
        for field in fields:
            field_value = obj.__dict__[field.name]
            field_type = types[field.name]
            self.serialize_any(field_value, field_type)
        self.decrease_container_depth()
def serialize_bool(self, value: bool)
Expand source code
def serialize_bool(self, value: st.bool):
    self.output.write(int(value).to_bytes(1, "little", signed=False))
def serialize_bytes(self, value: bytes)
Expand source code
def serialize_bytes(self, value: bytes):
    self.serialize_len(len(value))
    self.output.write(value)
def serialize_char(self, value: char)
Expand source code
def serialize_char(self, value: st.char):
    raise NotImplementedError
def serialize_f32(self, value: numpy.float32)
Expand source code
def serialize_f32(self, value: st.float32):
    raise NotImplementedError
def serialize_f64(self, value: numpy.float64)
Expand source code
def serialize_f64(self, value: st.float64):
    raise NotImplementedError
def serialize_i128(self, value: uint128)
Expand source code
def serialize_i128(self, value: st.uint128):
    self.output.write(int(value).to_bytes(16, "little", signed=True))
def serialize_i16(self, value: numpy.uint16)
Expand source code
def serialize_i16(self, value: st.uint16):
    self.output.write(int(value).to_bytes(2, "little", signed=True))
def serialize_i32(self, value: numpy.uint32)
Expand source code
def serialize_i32(self, value: st.uint32):
    self.output.write(int(value).to_bytes(4, "little", signed=True))
def serialize_i64(self, value: numpy.uint64)
Expand source code
def serialize_i64(self, value: st.uint64):
    self.output.write(int(value).to_bytes(8, "little", signed=True))
def serialize_i8(self, value: numpy.uint8)
Expand source code
def serialize_i8(self, value: st.uint8):
    self.output.write(int(value).to_bytes(1, "little", signed=True))
def serialize_len(self, value: int)
Expand source code
def serialize_len(self, value: int):
    raise NotImplementedError
def serialize_str(self, value: str)
Expand source code
def serialize_str(self, value: str):
    self.serialize_bytes(value.encode())
def serialize_u128(self, value: uint128)
Expand source code
def serialize_u128(self, value: st.uint128):
    self.output.write(int(value).to_bytes(16, "little", signed=False))
def serialize_u16(self, value: numpy.uint16)
Expand source code
def serialize_u16(self, value: st.uint16):
    self.output.write(int(value).to_bytes(2, "little", signed=False))
def serialize_u32(self, value: numpy.uint32)
Expand source code
def serialize_u32(self, value: st.uint32):
    self.output.write(int(value).to_bytes(4, "little", signed=False))
def serialize_u64(self, value: numpy.uint64)
Expand source code
def serialize_u64(self, value: st.uint64):
    self.output.write(int(value).to_bytes(8, "little", signed=False))
def serialize_u8(self, value: numpy.uint8)
Expand source code
def serialize_u8(self, value: st.uint8):
    self.output.write(int(value).to_bytes(1, "little", signed=False))
def serialize_unit(self, value: Type[NoneType])
Expand source code
def serialize_unit(self, value: st.unit):
    pass
def serialize_variant_index(self, value: int)
Expand source code
def serialize_variant_index(self, value: int):
    raise NotImplementedError
def sort_map_entries(self, offsets: List[int])
Expand source code
def sort_map_entries(self, offsets: typing.List[int]):
    raise NotImplementedError