Allow typing.Self as a hint-type.

README.md

@@ -158,6 +158,9 @@ All other types fall into the "complex" category.  They currently consist of:
 - `unions`: Unions of serializable types are supported as well.
 - `Structured`-derived types: You can use any of your `Structured`-derived classes as a type-hint,
   and the variable will be serialized as well.
+- `typing.Self`: This type-hint denotes that the attribute should be unpacked as an instance of
+  the containing class itself.  Note that due to the recursive posibilities this allows, care
+  must be taken to avoid hitting the recursion limit of Python.
 
 
 ### Tuples

structured/serializers/__init__.py

@@ -1,5 +1,6 @@
 from .api import *
 from .arrays import *
+from .self import *
 from .strings import *
 from .structs import *
 from .structured import *

structured/serializers/api.py

@@ -10,6 +10,17 @@
  - Modified packing method `pack`
  - All unpacking methods may return an iterable of values instead of a tuple.
 For more details, check the docstrings on each method or attribute.
+
+A note on "container" serializers (for example, CompoundSerializer and
+ArraySerializer): Due to the posibility of recursive nesting via the
+`typing.Self` type-hint as a serializable type, care must be taken with
+delegating to sub-serializers. In particular, only updating `self.size` at the
+*end* of a pack/unpack operation ensures that nested usages of the same
+serializer won't overwrite intermediate values.
+
+Similarly (although this is true regardless of nesting), you almost always want
+a custom `prepack` and `preunpack` method, to pass that information along to
+the nested serializers.
 """
 
 from __future__ import annotations

structured/serializers/arrays.py

@@ -97,25 +97,37 @@ def _check_data_size(self, expected: int, actual: int) -> None:
                 f'Array data size {actual} does not match expected size {expected}'
             )
 
+    def prepack(self, partial_object) -> Self:
+        self._partial_object = partial_object
+        return self
+
+    def preunpack(self, partial_object) -> Self:
+        self._partial_object = partial_object
+        return self
+
     def pack(self, *values: Unpack[tuple[list[T]]]) -> bytes:
         data = [b'']
-        self.size = header_size = self.header_serializer.size
+        size = header_size = self.header_serializer.size
+        item_serializer = self.item_serializer.prepack(self._partial_object)
         for item in values[0]:
-            data.append(self.item_serializer.pack(item))
-            self.size += self.item_serializer.size
-        header_values = self._header_pack_values(values[0], self.size - header_size)
+            data.append(item_serializer.pack(item))
+            size += item_serializer.size
+        header_values = self._header_pack_values(values[0], size - header_size)
         data[0] = self.header_serializer.pack(*header_values)
+        self.size = size
         return b''.join(data)
 
     def pack_into(
         self, buffer: WritableBuffer, offset: int, *values: Unpack[tuple[list[T]]]
     ) -> None:
         items = values[0]
-        self.size = header_size = self.header_serializer.size
+        size = header_size = self.header_serializer.size
+        item_serializer = self.item_serializer.prepack(self._partial_object)
         for item in items:
-            self.item_serializer.pack_into(buffer, offset + self.size, item)
-            self.size += self.item_serializer.size
-        header_values = self._header_pack_values(items, self.size - header_size)
+            item_serializer.pack_into(buffer, offset + size, item)
+            size += item_serializer.size
+        header_values = self._header_pack_values(items, size - header_size)
+        self.size = size
         self.header_serializer.pack_into(buffer, offset, *header_values)
 
     def pack_write(self, writable: BinaryIO, *values: Unpack[tuple[list[T]]]) -> None:
@@ -125,34 +137,40 @@ def pack_write(self, writable: BinaryIO, *values: Unpack[tuple[list[T]]]) -> Non
     def unpack(self, buffer: ReadableBuffer) -> tuple[list[T]]:
         header = self.header_serializer.unpack(buffer)
         count, data_size = self._header_unpack_values(*header)
-        self.size = header_size = self.header_serializer.size
+        size = header_size = self.header_serializer.size
+        item_serializer = self.item_serializer.preunpack(self._partial_object)
         items = []
         for _ in range(count):
-            items.extend(self.item_serializer.unpack(buffer[self.size :]))
-            self.size += self.item_serializer.size
-        self._check_data_size(data_size, self.size - header_size)
+            items.extend(item_serializer.unpack(buffer[size:]))
+            size += item_serializer.size
+        self._check_data_size(data_size, size - header_size)
+        self.size = size
         return (items,)
 
     def unpack_from(self, buffer: ReadableBuffer, offset: int) -> tuple[list[T]]:
         header = self.header_serializer.unpack_from(buffer, offset)
         count, data_size = self._header_unpack_values(*header)
-        self.size = header_size = self.header_serializer.size
+        size = header_size = self.header_serializer.size
+        item_serializer = self.item_serializer.preunpack(self._partial_object)
         items = []
         for _ in range(count):
-            items.extend(self.item_serializer.unpack_from(buffer, offset + self.size))
-            self.size += self.item_serializer.size
-        self._check_data_size(data_size, self.size - header_size)
+            items.extend(item_serializer.unpack_from(buffer, offset + size))
+            size += item_serializer.size
+        self._check_data_size(data_size, size - header_size)
+        self.size = size
         return (items,)
 
     def unpack_read(self, readable: BinaryIO) -> tuple[list[T]]:
         header = self.header_serializer.unpack_read(readable)
         count, data_size = self._header_unpack_values(*header)
-        self.size = header_size = self.header_serializer.size
+        size = header_size = self.header_serializer.size
+        item_serializer = self.item_serializer.preunpack(self._partial_object)
         items = []
         for _ in range(count):
-            items.extend(self.item_serializer.unpack_read(readable))
-            self.size += self.item_serializer.size
-        self._check_data_size(data_size, self.size - header_size)
+            items.extend(item_serializer.unpack_read(readable))
+            size += item_serializer.size
+        self._check_data_size(data_size, size - header_size)
+        self.size = size
         return (items,)
 
 

structured/serializers/self.py

@@ -0,0 +1,36 @@
+"""
+Serializer for special handling of the typing.Self typehint.
+"""
+
+__all__ = [
+    'SelfSerializer',
+]
+
+
+from ..type_checking import TYPE_CHECKING, Any, ClassVar, Self, annotated
+from .api import Serializer
+from .structured import StructuredSerializer
+
+if TYPE_CHECKING:
+    from ..structured import Structured, _Proxy
+else:
+    Structured = 'Structured'
+    _Proxy = '_Proxy'
+
+
+class SelfSerializer(Serializer[Structured]):
+    num_values: ClassVar[int] = 1
+
+    def prepack(self, partial_object: Structured) -> Serializer:
+        return StructuredSerializer(type(partial_object))
+
+    def preunpack(self, partial_object: _Proxy) -> Serializer:
+        return StructuredSerializer(partial_object.cls)
+
+    @classmethod
+    def _transform(cls, unwrapped: Any, actual: Any) -> Any:
+        if unwrapped is Self:
+            return cls()
+
+
+annotated.register_transform(SelfSerializer._transform)

structured/serializers/structured.py

@@ -42,32 +42,39 @@ class StructuredSerializer(Generic[TStructured], Serializer[TStructured]):
 
     def __init__(self, obj_type: type[TStructured]) -> None:
         self.obj_type = obj_type
-
-    @property
-    def size(self) -> int:
-        return self.obj_type.serializer.size
+        self.size = 0
 
     def pack(self, values: TStructured) -> bytes:
-        return values.pack()
+        data = values.pack()
+        self.size = values.serializer.size
+        return data
 
     def pack_into(
         self, buffer: WritableBuffer, offset: int, values: TStructured
     ) -> None:
         values.pack_into(buffer, offset)
+        self.size = values.serializer.size
 
     def pack_write(self, writable: BinaryIO, values: TStructured) -> None:
         values.pack_write(writable)
+        self.size = values.serializer.size
 
     def unpack(self, buffer: ReadableBuffer) -> tuple[TStructured]:
-        return (self.obj_type.create_unpack(buffer),)
+        value = self.obj_type.create_unpack(buffer)
+        self.size = self.obj_type.serializer.size
+        return (value,)
 
     def unpack_from(
         self, buffer: ReadableBuffer, offset: int = 0
     ) -> tuple[TStructured]:
-        return (self.obj_type.create_unpack_from(buffer, offset),)
+        value = self.obj_type.create_unpack_from(buffer, offset)
+        self.size = self.obj_type.serializer.size
+        return (value,)
 
     def unpack_read(self, readable: BinaryIO) -> tuple[TStructured]:
-        return (self.obj_type.create_unpack_read(readable),)
+        value = self.obj_type.create_unpack_read(readable)
+        self.size = self.obj_type.serializer.size
+        return (value,)
 
     @classmethod
     def _transform(cls, unwrapped: Any, actual: Any) -> Any:

structured/serializers/unions.py

@@ -20,6 +20,7 @@
     ClassVar,
     Iterable,
     ReadableBuffer,
+    WritableBuffer,
     annotated,
     get_union_args,
 )
@@ -48,7 +49,7 @@ def __init__(self, result_map: dict[Any, Any], default: Any = None) -> None:
             key: self.validate_serializer(serializer)
             for key, serializer in result_map.items()
         }
-        self._last_serializer = self.default
+        self.size = 0
 
     @staticmethod
     def validate_serializer(hint) -> Serializer:
@@ -59,16 +60,15 @@ def validate_serializer(hint) -> Serializer:
             raise ValueError('Union results must serializer a single item.')
         return serializer
 
-    @property
-    def size(self) -> int:
-        if self._last_serializer:
-            return self._last_serializer.size
-        else:
-            return 0
+    def prepack(self, partial_object) -> Serializer:
+        self._partial_object = partial_object
+        return self
+
+    def preunpack(self, partial_object) -> Serializer:
+        self._partial_object = partial_object
+        return self
 
-    def get_serializer(
-        self, decider_result: Any, partial_object: Any, packing: bool
-    ) -> Serializer:
+    def get_serializer(self, decider_result: Any, packing: bool) -> Serializer:
         """Given a target used to decide, return a serializer used to unpack."""
         if self.default is None:
             try:
@@ -80,11 +80,9 @@ def get_serializer(
         else:
             serializer = self.result_map.get(decider_result, self.default)
         if packing:
-            serializer = serializer.prepack(partial_object)
+            return serializer.prepack(self._partial_object)
         else:
-            serializer = serializer.preunpack(partial_object)
-        self._last_serializer = serializer
-        return self._last_serializer
+            return serializer.preunpack(self._partial_object)
 
     @staticmethod
     def _transform(unwrapped: Any, actual: Any) -> Any:
@@ -120,13 +118,43 @@ def __init__(
         super().__init__(result_map, default)
         self.decider = decider
 
-    def prepack(self, partial_object: Any) -> Serializer:
-        result = self.decider(partial_object)
-        return self.get_serializer(result, partial_object, True)
+    def decide(self, packing: bool) -> Serializer:
+        result = self.decider(self._partial_object)
+        return self.get_serializer(result, packing)
+
+    def pack(self, *values: Any) -> bytes:
+        serializer = self.decide(True)
+        data = serializer.pack(*values)
+        self.size = serializer.size
+        return data
+
+    def pack_into(self, buffer: WritableBuffer, offset: int, *values: Any) -> None:
+        serializer = self.decide(True)
+        serializer.pack_into(buffer, offset, *values)
+        self.size = serializer.size
+
+    def pack_write(self, writable: BinaryIO, *values: Any) -> None:
+        serializer = self.decide(True)
+        serializer.pack_write(writable, *values)
+        self.size = serializer.size
 
-    def preunpack(self, partial_object: Any) -> Serializer:
-        result = self.decider(partial_object)
-        return self.get_serializer(result, partial_object, False)
+    def unpack(self, buffer: ReadableBuffer) -> Iterable:
+        serializer = self.decide(False)
+        value = serializer.unpack(buffer)
+        self.size = serializer.size
+        return value
+
+    def unpack_from(self, buffer: ReadableBuffer, offset: int = 0) -> Iterable:
+        serializer = self.decide(False)
+        value = serializer.unpack_from(buffer, offset)
+        self.size = serializer.size
+        return value
+
+    def unpack_read(self, readable: BinaryIO) -> Iterable:
+        serializer = self.decide(False)
+        value = serializer.unpack_read(readable)
+        self.size = serializer.size
+        return value
 
 
 class LookaheadDecider(AUnion):
@@ -153,19 +181,43 @@ def __init__(
             )
         self.read_ahead_serializer = serializer
 
-    def prepack(self, partial_object: Any) -> Serializer:
-        result = self.decider(partial_object)
-        return self.get_serializer(result, partial_object, True)
+    def pack(self, *values: Any) -> bytes:
+        result = self.decider(self._partial_object)
+        serializer = self.get_serializer(result, True)
+        data = serializer.pack(*values)
+        self.size = serializer.size
+        return data
+
+    def pack_into(self, buffer: WritableBuffer, offset: int, *values: Any) -> None:
+        result = self.decider(self._partial_object)
+        serializer = self.get_serializer(result, True)
+        serializer.pack_into(buffer, offset, *values)
+        self.size = serializer.size
+
+    def pack_write(self, writable: BinaryIO, *values: Any) -> None:
+        result = self.decider(self._partial_object)
+        serializer = self.get_serializer(result, True)
+        serializer.pack_write(writable, *values)
+        self.size = serializer.size
 
     def unpack(self, buffer: ReadableBuffer) -> Iterable:
         result = tuple(self.read_ahead_serializer.unpack(buffer))[0]
-        return self.get_serializer(result, None, False).unpack(buffer)
+        serializer = self.get_serializer(result, False)
+        values = serializer.unpack(buffer)
+        self.size = serializer.size
+        return values
 
     def unpack_from(self, buffer: ReadableBuffer, offset: int = 0) -> Iterable:
         result = tuple(self.read_ahead_serializer.unpack_from(buffer, offset))[0]
-        return self.get_serializer(result, None, False).unpack_from(buffer, offset)
+        serializer = self.get_serializer(result, False)
+        values = serializer.unpack_from(buffer, offset)
+        self.size = serializer.size
+        return values
 
     def unpack_read(self, readable: BinaryIO) -> Iterable:
         result = tuple(self.read_ahead_serializer.unpack_read(readable))[0]
         readable.seek(-self.read_ahead_serializer.size, os.SEEK_CUR)
-        return self.get_serializer(result, None, False).unpack_read(readable)
+        serializer = self.get_serializer(result, False)
+        values = serializer.unpack_read(readable)
+        self.size = serializer.size
+        return values

structured/structured.py

@@ -285,7 +285,7 @@ def _create_proxy(cls) -> tuple[_Proxy, Serializer]:
         """Create a proxy object for this class, which can be used to create
         new instances of this class.
         """
-        proxy = _Proxy(cls.attrs)
+        proxy = _Proxy(cls)
         return proxy, cls.serializer.preunpack(proxy)
 
     @classmethod
@@ -464,8 +464,9 @@ class _Proxy:
 
     # NOTE: Only using __dunder__ methods, so any attributes on the class this
     # is a proxy for won't be shadowed.
-    def __init__(self, attrs: tuple[str, ...]) -> None:
-        self.__attrs = attrs
+    def __init__(self, cls: type[Structured]) -> None:
+        self.__attrs = cls.attrs
+        self.cls = cls
 
     def __call__(self, values: Iterable[Any]) -> None:
         for attr, value in zips(self.__attrs, values, strict=True):