Composite Serializers

Packet serializer composite module.

final class easynetwork.serializers.composite.StapledPacketSerializer

Bases: AbstractPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket]

A composite serializer that merges two serializers.

See also

How-to — Serializer Combinations: This class cannot be used directly with a StreamProtocol. This page explains possible workarounds.

property sent_packet_serializer: AbstractPacketSerializer[_T_SentDTOPacket, Any]: Sent packet serializer.

property received_packet_serializer: AbstractPacketSerializer[Any, _T_ReceivedDTOPacket]: Received packet serializer.

serialize(packet: _T_SentDTOPacket) → bytes

Calls self.sent_packet_serializer.serialize(packet).

Parameters:: packet (_T_SentDTOPacket) – The Python object to serialize.
Returns:: a byte sequence.
Return type:: bytes

deserialize(data: bytes) → _T_ReceivedDTOPacket

Calls self.received_packet_serializer.deserialize(data).

Parameters:: data (bytes) – The byte sequence to deserialize.
Raises:: DeserializeError – An unrelated deserialization error occurred.
Returns:: the deserialized Python object.
Return type:: _T_ReceivedDTOPacket

final class easynetwork.serializers.composite.StapledIncrementalPacketSerializer

Bases: StapledPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket], AbstractIncrementalPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket], Generic[_T_SentDTOPacket, _T_ReceivedDTOPacket]

A composite serializer that merges two incremental serializers.

property sent_packet_serializer: AbstractPacketSerializer[_T_SentDTOPacket, Any]: Sent packet serializer.

property received_packet_serializer: AbstractPacketSerializer[Any, _T_ReceivedDTOPacket]: Received packet serializer.

incremental_serialize(packet: _T_SentDTOPacket) → Generator[bytes]

Calls self.sent_packet_serializer.incremental_serialize(packet).

Parameters:: packet (_T_SentDTOPacket) – The Python object to serialize.
Yields:: all the parts of the packet.
Return type:: Generator[bytes]

incremental_deserialize() → Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

Calls self.received_packet_serializer.incremental_deserialize().

Raises:: IncrementalDeserializeError – An unrelated deserialization error occurred.
Yields:: None until the whole packet has been deserialized.
Returns:: a tuple with the deserialized Python object and the unused trailing data.
Return type:: Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

deserialize(data: bytes) → _T_ReceivedDTOPacket

Calls self.received_packet_serializer.deserialize(data).

Parameters:: data (bytes) – The byte sequence to deserialize.
Raises:: DeserializeError – An unrelated deserialization error occurred.
Returns:: the deserialized Python object.
Return type:: _T_ReceivedDTOPacket

serialize(packet: _T_SentDTOPacket) → bytes

Calls self.sent_packet_serializer.serialize(packet).

Parameters:: packet (_T_SentDTOPacket) – The Python object to serialize.
Returns:: a byte sequence.
Return type:: bytes

final class easynetwork.serializers.composite.StapledBufferedIncrementalPacketSerializer

Bases: StapledIncrementalPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket], BufferedIncrementalPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket, _T_Buffer], Generic[_T_SentDTOPacket, _T_ReceivedDTOPacket, _T_Buffer]

A composite serializer that merges two incremental serializers with manual control of the receive buffer.

deserialize(data: bytes) → _T_ReceivedDTOPacket

Calls self.received_packet_serializer.deserialize(data).

Parameters:: data (bytes) – The byte sequence to deserialize.
Raises:: DeserializeError – An unrelated deserialization error occurred.
Returns:: the deserialized Python object.
Return type:: _T_ReceivedDTOPacket

incremental_deserialize() → Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

Calls self.received_packet_serializer.incremental_deserialize().

Raises:: IncrementalDeserializeError – An unrelated deserialization error occurred.
Yields:: None until the whole packet has been deserialized.
Returns:: a tuple with the deserialized Python object and the unused trailing data.
Return type:: Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

incremental_serialize(packet: _T_SentDTOPacket) → Generator[bytes]

Calls self.sent_packet_serializer.incremental_serialize(packet).

Parameters:: packet (_T_SentDTOPacket) – The Python object to serialize.
Yields:: all the parts of the packet.
Return type:: Generator[bytes]

serialize(packet: _T_SentDTOPacket) → bytes

Calls self.sent_packet_serializer.serialize(packet).

Parameters:: packet (_T_SentDTOPacket) – The Python object to serialize.
Returns:: a byte sequence.
Return type:: bytes

property sent_packet_serializer: AbstractPacketSerializer[_T_SentDTOPacket, Any]: Sent packet serializer.

property received_packet_serializer: AbstractPacketSerializer[Any, _T_ReceivedDTOPacket]: Received packet serializer.

create_deserializer_buffer(sizehint: int) → _T_Buffer

Calls self.received_packet_serializer.create_deserializer_buffer(sizehint).

Parameters:: sizehint (int) – the recommended size (in bytes) for the returned buffer. It is acceptable to return smaller or larger buffers than what sizehint suggests.
Returns:: an object implementing the buffer protocol. It is an error to return a buffer with a zero size.
Return type:: _T_Buffer

buffered_incremental_deserialize(buffer: _T_Buffer) → Generator[int | None, int, tuple[_T_ReceivedDTOPacket, TypeAliasForwardRef('bytes | bytearray | memoryview | collections.abc.Buffer')]]

Calls self.received_packet_serializer.buffered_incremental_deserialize(buffer).

Parameters:

buffer (_T_Buffer) – The buffer allocated by create_deserializer_buffer().

Raises:

IncrementalDeserializeError – An unrelated deserialization error occurred.

Yields:

until the whole packet has been deserialized.

Returns:

a tuple with the deserialized Python object and the unused trailing data.

The remainder can be a memoryview pointing to buffer or an external bytes-like object.

Return type:

Generator[int | None, int, tuple[_T_ReceivedDTOPacket, TypeAliasForwardRef(‘bytes | bytearray | memoryview | collections.abc.Buffer’)]]