feat(nano): implement serialization module [part 1]

hathor/serialization/__init__.py

@@ -0,0 +1,27 @@
+# Copyright 2025 Hathor Labs
+#
+# 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.
+
+from .deserializer import Deserializer
+from .exceptions import BadDataError, OutOfDataError, SerializationError, TooLongError, UnsupportedTypeError
+from .serializer import Serializer
+
+__all__ = [
+    'Serializer',
+    'Deserializer',
+    'SerializationError',
+    'UnsupportedTypeError',
+    'TooLongError',
+    'OutOfDataError',
+    'BadDataError',
+]

hathor/serialization/adapters/__init__.py

@@ -0,0 +1,24 @@
+# Copyright 2025 Hathor Labs
+#
+# 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.
+
+from .generic_adapter import GenericDeserializerAdapter, GenericSerializerAdapter
+from .max_bytes import MaxBytesDeserializer, MaxBytesExceededError, MaxBytesSerializer
+
+__all__ = [
+    'GenericDeserializerAdapter',
+    'GenericSerializerAdapter',
+    'MaxBytesDeserializer',
+    'MaxBytesExceededError',
+    'MaxBytesSerializer',
+]

hathor/serialization/adapters/generic_adapter.py

@@ -0,0 +1,110 @@
+# Copyright 2025 Hathor Labs
+#
+# 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.
+
+from types import TracebackType
+from typing import Generic, TypeVar
+
+from typing_extensions import Self, override
+
+from hathor.serialization.deserializer import Deserializer
+from hathor.serialization.serializer import Serializer
+
+from ..types import Buffer
+
+S = TypeVar('S', bound=Serializer)
+D = TypeVar('D', bound=Deserializer)
+
+
+class GenericSerializerAdapter(Serializer, Generic[S]):
+    inner: S
+
+    def __init__(self, serializer: S) -> None:
+        self.inner = serializer
+
+    @override
+    def finalize(self) -> Buffer:
+        return self.inner.finalize()
+
+    @override
+    def cur_pos(self) -> int:
+        return self.inner.cur_pos()
+
+    @override
+    def write_byte(self, data: int) -> None:
+        self.inner.write_byte(data)
+
+    @override
+    def write_bytes(self, data: Buffer) -> None:
+        self.inner.write_bytes(data)
+
+    # allow using this adapter as a context manager:
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        pass
+
+
+class GenericDeserializerAdapter(Deserializer, Generic[D]):
+    inner: D
+
+    def __init__(self, deserializer: D) -> None:
+        self.inner = deserializer
+
+    @override
+    def finalize(self) -> None:
+        return self.inner.finalize()
+
+    @override
+    def is_empty(self) -> bool:
+        return self.inner.is_empty()
+
+    @override
+    def peek_byte(self) -> int:
+        return self.inner.peek_byte()
+
+    @override
+    def peek_bytes(self, n: int, *, exact: bool = True) -> Buffer:
+        return self.inner.peek_bytes(n, exact=exact)
+
+    @override
+    def read_byte(self) -> int:
+        return self.inner.read_byte()
+
+    @override
+    def read_bytes(self, n: int, *, exact: bool = True) -> Buffer:
+        return self.inner.read_bytes(n, exact=exact)
+
+    @override
+    def read_all(self) -> Buffer:
+        return self.inner.read_all()
+
+    # allow using this adapter as a context manager:
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        pass

hathor/serialization/adapters/max_bytes.py

@@ -0,0 +1,91 @@
+# Copyright 2025 Hathor Labs
+#
+# 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.
+
+from typing import TypeVar
+
+from typing_extensions import override
+
+from hathor.serialization.deserializer import Deserializer
+from hathor.serialization.exceptions import SerializationError
+from hathor.serialization.serializer import Serializer
+
+from ..types import Buffer
+from .generic_adapter import GenericDeserializerAdapter, GenericSerializerAdapter
+
+S = TypeVar('S', bound=Serializer)
+D = TypeVar('D', bound=Deserializer)
+
+
+class MaxBytesExceededError(SerializationError):
+    """ This error is raised when the adapted serializer reached its maximum bytes write/read.
+
+    After this exception is raised the adapted serializer cannot be used anymore. Handlers of this exception are
+    expected to either: bubble up the exception (or an equivalente exception), or return an error. Handlers should not
+    try to write again on the same serializer.
+
+    It is possible that the inner serializer is still usable, but the point where the serialized stopped writing or
+    reading might leave the rest of the data unusable, so for that reason it should be considered a failed
+    (de)serialization overall, and not simply a failed "read/write" operation.
+    """
+    pass
+
+
+class MaxBytesSerializer(GenericSerializerAdapter[S]):
+    def __init__(self, serializer: S, max_bytes: int) -> None:
+        super().__init__(serializer)
+        self._bytes_left = max_bytes
+
+    def _check_update_exceeds(self, write_size: int) -> None:
+        self._bytes_left -= write_size
+        if self._bytes_left < 0:
+            raise MaxBytesExceededError
+
+    @override
+    def write_byte(self, data: int) -> None:
+        self._check_update_exceeds(1)
+        super().write_byte(data)
+
+    @override
+    def write_bytes(self, data: Buffer) -> None:
+        data_view = memoryview(data)
+        self._check_update_exceeds(len(data_view))
+        super().write_bytes(data_view)
+
+
+class MaxBytesDeserializer(GenericDeserializerAdapter[D]):
+    def __init__(self, deserializer: D, max_bytes: int) -> None:
+        super().__init__(deserializer)
+        self._bytes_left = max_bytes
+
+    def _check_update_exceeds(self, read_size: int) -> None:
+        self._bytes_left -= read_size
+        if self._bytes_left < 0:
+            raise MaxBytesExceededError
+
+    @override
+    def read_byte(self) -> int:
+        self._check_update_exceeds(1)
+        return super().read_byte()
+
+    @override
+    def read_bytes(self, n: int, *, exact: bool = True) -> Buffer:
+        self._check_update_exceeds(n)
+        return super().read_bytes(n, exact=exact)
+
+    @override
+    def read_all(self) -> Buffer:
+        result = super().read_bytes(self._bytes_left, exact=False)
+        if not self.is_empty():
+            raise MaxBytesExceededError
+        return result

hathor/serialization/bytes_deserializer.py

@@ -0,0 +1,76 @@
+# Copyright 2025 Hathor Labs
+#
+# 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.
+
+from typing_extensions import override
+
+from .deserializer import Deserializer
+from .exceptions import OutOfDataError
+from .types import Buffer
+
+_EMPTY_VIEW = memoryview(b'')
+
+
+class BytesDeserializer(Deserializer):
+    """Simple implementation of a Deserializer to parse values from a byte sequence.
+
+    This implementation maintains a memoryview that is shortened as the bytes are read.
+    """
+
+    def __init__(self, data: Buffer) -> None:
+        self._view = memoryview(data)
+
+    @override
+    def finalize(self) -> None:
+        if not self.is_empty():
+            raise ValueError('trailing data')
+        del self._view
+
+    @override
+    def is_empty(self) -> bool:
+        # XXX: least amount of OPs, "not" converts to bool with the correct semantics of "is empty"
+        return not self._view
+
+    @override
+    def peek_byte(self) -> int:
+        if not len(self._view):
+            raise OutOfDataError('not enough bytes to read')
+        return self._view[0]
+
+    @override
+    def peek_bytes(self, n: int, *, exact: bool = True) -> memoryview:
+        if n < 0:
+            raise ValueError('value cannot be negative')
+        if exact and len(self._view) < n:
+            raise OutOfDataError('not enough bytes to read')
+        return self._view[:n]
+
+    @override
+    def read_byte(self) -> int:
+        b = self.peek_byte()
+        self._view = self._view[1:]
+        return b
+
+    @override
+    def read_bytes(self, n: int, *, exact: bool = True) -> memoryview:
+        b = self.peek_bytes(n, exact=exact)
+        if exact and len(self._view) < n:
+            raise OutOfDataError('not enough bytes to read')
+        self._view = self._view[n:]
+        return b
+
+    @override
+    def read_all(self) -> memoryview:
+        b = self._view
+        self._view = _EMPTY_VIEW
+        return b

hathor/serialization/bytes_serializer.py

@@ -0,0 +1,53 @@
+# Copyright 2025 Hathor Labs
+#
+# 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.
+
+from typing_extensions import override
+
+from .serializer import Serializer
+from .types import Buffer
+
+
+class BytesSerializer(Serializer):
+    """Simple implementation of Serializer to write to memory.
+
+    This implementation defers joining everything until finalize is called, before that every write is stored as a
+    memoryview in a list.
+    """
+
+    def __init__(self) -> None:
+        self._parts: list[memoryview] = []
+        self._pos: int = 0
+
+    @override
+    def finalize(self) -> memoryview:
+        result = memoryview(b''.join(self._parts))
+        del self._parts
+        del self._pos
+        return result
+
+    @override
+    def cur_pos(self) -> int:
+        return self._pos
+
+    @override
+    def write_byte(self, data: int) -> None:
+        # int.to_bytes checks for correct range
+        self._parts.append(memoryview(int.to_bytes(data)))
+        self._pos += 1
+
+    @override
+    def write_bytes(self, data: Buffer) -> None:
+        part = memoryview(data)
+        self._parts.append(part)
+        self._pos += len(part)