Abstract Base Classes

---

Top-Level Base Classes

Network packet serializer base classes module.

class easynetwork.serializers.abc.AbstractIncrementalPacketSerializer

Bases: AbstractPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket]

The base class for implementing an incremental serializer.

abstractmethod incremental_serialize(packet: _T_SentDTOPacket, /) → Generator[bytes]

Returns the byte representation of the Python object packet.

The generator should yield non-empty byte sequences.

The main purpose of this method is to add metadata that could not be included in the output of serialize(), such as headers, separators, and so on. It is used in the incremental_deserialize() method.

Parameters:

packet (_T_SentDTOPacket) – The Python object to serialize.

Yields:

all the parts of the packet.

Return type:

Generator[bytes]

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

Creates a Python object representing the raw packet.

Raises:

IncrementalDeserializeError – An unrelated deserialization error occurred.

Yields:

None until the whole packet has been deserialized.

At each yield checkpoint, the endpoint implementation sends to the generator the data received from the remote endpoint.

Returns:

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

Return type:

Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

serialize(packet: _T_SentDTOPacket, /) → bytes

Returns the byte representation of the Python object packet.

The default implementation concatenates and returns the parts sent by incremental_serialize().

Parameters:

packet (_T_SentDTOPacket) – The Python object to serialize.

Returns:

a byte sequence.

Return type:

bytes

deserialize(data: bytes, /) → _T_ReceivedDTOPacket

Creates a Python object representing the raw packet from data.

The default implementation uses incremental_deserialize() and expects it to deserialize data at once.

Parameters:

data (bytes) – The byte sequence to deserialize.

Raises:

• DeserializeError – Too little or too much data to parse.

• DeserializeError – An unrelated deserialization error occurred.

Returns:

the deserialized Python object.

Return type:

_T_ReceivedDTOPacket

class easynetwork.serializers.abc.AbstractPacketSerializer

Bases: Generic[_T_SentDTOPacket, _T_ReceivedDTOPacket]

The base class for implementing a serializer.

Implementing this interface would create a one-shot serializer.

See also

How-to — Serializer Combinations

This class cannot be used directly with a StreamProtocol. This page explains possible workarounds.

abstractmethod serialize(packet: _T_SentDTOPacket, /) → bytes

Returns the byte representation of the Python object packet.

Parameters:

packet (_T_SentDTOPacket) – The Python object to serialize.

Returns:

a byte sequence.

Return type:

bytes

abstractmethod deserialize(data: bytes, /) → _T_ReceivedDTOPacket

Creates a Python object representing the raw packet from 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

class easynetwork.serializers.abc.BufferedIncrementalPacketSerializer

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

The base class for implementing an incremental serializer with manual control of the receive buffer.

abstractmethod create_deserializer_buffer(sizehint: int, /) → _T_Buffer

Called to allocate a new receive buffer.

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

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

Creates a Python object representing the raw packet.

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.

The value yielded is the position to start writing to the buffer. It can be:

• None: Just use the whole buffer. Therefore, yield and yield None are equivalent to yield 0.

• A positive integer (starting at 0): Skips the first n bytes.

• A negative integer: Skips until the last n bytes. For example, yield -10 means to write from the last 10th byte of the buffer.

At each yield checkpoint, the endpoint implementation sends to the generator the number of bytes written to the buffer.

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’)]]

---

Stream Base Classes

Stream network packet serializer handler module.

Here are abstract classes that implement common stream protocol patterns.

class easynetwork.serializers.base_stream.AutoSeparatedPacketSerializer

Bases: BufferedIncrementalPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket, bytearray]

Base class for stream protocols that separates sent information by a byte sequence.

__init__(separator: bytes, *, incremental_serialize_check_separator: bool = True, limit: int = DEFAULT_SERIALIZER_LIMIT, debug: bool = False, **kwargs: Any) → None

Parameters:

• separator (bytes) – Byte sequence that indicates the end of the token.

• incremental_serialize_check_separator (bool) – If True (the default), checks that the data returned by serialize() does not contain separator, and removes superfluous separator added at the end.

• limit (int) – Maximum buffer size.

• debug (bool) – If True, add information to DeserializeError via the error_info attribute.

• kwargs (Any) – Extra options given to super().__init__().

Raises:

• TypeError – Invalid arguments.

• ValueError – Empty separator sequence.

• ValueError – limit must be a positive integer.

abstractmethod serialize(packet: _T_SentDTOPacket, /) → bytes

See AbstractPacketSerializer.serialize() documentation.

Return type:

bytes

final incremental_serialize(packet: _T_SentDTOPacket, /) → Generator[bytes]

Yields the data returned by serialize() and appends separator.

See AbstractIncrementalPacketSerializer.incremental_serialize() documentation for details.

Raises:

• ValueError – If incremental_serialize_check_separator is True and separator is in the returned data.

• Exception – Any error raised by serialize().

Return type:

Generator[bytes]

abstractmethod deserialize(data: bytes, /) → _T_ReceivedDTOPacket

See AbstractPacketSerializer.deserialize() documentation.

Return type:

_T_ReceivedDTOPacket

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

Yields until separator is found and calls deserialize() without separator.

See AbstractIncrementalPacketSerializer.incremental_deserialize() documentation for details.

Raises:

• LimitOverrunError – Reached buffer size limit.

• IncrementalDeserializeError – deserialize() raised DeserializeError.

• Exception – Any error raised by deserialize().

Return type:

Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

final create_deserializer_buffer(sizehint: int) → bytearray

See BufferedIncrementalPacketSerializer.buffered_incremental_deserialize() documentation for details.

Return type:

bytearray

final buffered_incremental_deserialize(buffer: bytearray) → Generator[int, int, tuple[_T_ReceivedDTOPacket, memoryview]]

Yields until separator is found and calls deserialize() without separator.

See BufferedIncrementalPacketSerializer.buffered_incremental_deserialize() documentation for details.

Raises:

• LimitOverrunError – Reached buffer size limit.

• IncrementalDeserializeError – deserialize() raised DeserializeError.

• Exception – Any error raised by deserialize().

Return type:

Generator[int, int, tuple[_T_ReceivedDTOPacket, memoryview]]

property separator: bytes

Byte sequence that indicates the end of the token. Read-only attribute.

property debug: bool

The debug mode flag. Read-only attribute.

property buffer_limit: int

Maximum buffer size. Read-only attribute.

class easynetwork.serializers.base_stream.FileBasedPacketSerializer

Bases: BufferedIncrementalPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket, memoryview]

Base class for APIs requiring a file object for serialization/deserialization.

__init__(expected_load_error: type[Exception] | tuple[type[Exception], ...], *, limit: int = DEFAULT_SERIALIZER_LIMIT, debug: bool = False, **kwargs: Any) → None

Parameters:

• expected_load_error (type[Exception] | tuple[type[Exception], ...]) – Errors that can be raised by load_from_file() implementation, which must be considered as deserialization errors.

• limit (int) – Maximum buffer size.

• debug (bool) – If True, add information to DeserializeError via the error_info attribute.

• kwargs (Any) – Extra options given to super().__init__().

abstractmethod dump_to_file(packet: _T_SentDTOPacket, file: IO[bytes], /) → None

Write the serialized packet to file.

Parameters:

• packet (_T_SentDTOPacket) – The Python object to serialize.

• file (IO[bytes]) – The binary file to write to.

abstractmethod load_from_file(file: IO[bytes], /) → _T_ReceivedDTOPacket

Read from file to deserialize the raw packet.

Parameters:

file (IO[bytes]) – The binary file to read from.

Raises:

• EOFError – Missing data to create the packet.

• Exception – Any error from the underlying API that is interpreted with the expected_load_error value.

Returns:

the deserialized Python object.

Return type:

_T_ReceivedDTOPacket

serialize(packet: _T_SentDTOPacket, /) → bytes

Returns the byte representation of the Python object packet.

By default, this method uses dump_to_file(), but it can be overriden for speicific usage.

See AbstractPacketSerializer.serialize() documentation for details.

Raises:

Exception – Any error raised by dump_to_bytes().

Return type:

bytes

deserialize(data: bytes, /) → _T_ReceivedDTOPacket

Creates a Python object representing the raw packet from data.

By default, this method uses load_from_file(), but it can be overriden for speicific usage.

See AbstractPacketSerializer.deserialize() documentation for details.

Raises:

• DeserializeError – load_from_file() raised EOFError.

• DeserializeError – load_from_file() does not read until EOF (unused trailing data).

• DeserializeError – load_from_file() raised an error that matches expected_load_error.

• Exception – Any other error raised by load_from_file().

Return type:

_T_ReceivedDTOPacket

final incremental_serialize(packet: _T_SentDTOPacket, /) → Generator[bytes]

Calls dump_to_file() and yields the result.

See AbstractIncrementalPacketSerializer.incremental_serialize() documentation for details.

Raises:

Exception – Any error raised by dump_to_file().

Return type:

Generator[bytes]

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

Calls load_from_file() and returns the result.

See AbstractIncrementalPacketSerializer.incremental_deserialize() documentation for details.

Note

The generator will always yield if load_from_file() raises EOFError.

Raises:

• IncrementalDeserializeError – load_from_file() raised an error that matches expected_load_error.

• Exception – Any other error raised by load_from_file().

Return type:

Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

final create_deserializer_buffer(sizehint: int, /) → memoryview

See BufferedIncrementalPacketSerializer.create_deserializer_buffer() documentation for details.

Return type:

memoryview

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

Calls load_from_file() and returns the result.

See BufferedIncrementalPacketSerializer.buffered_incremental_deserialize() documentation for details.

Note

The generator will always yield if load_from_file() raises EOFError.

Raises:

• IncrementalDeserializeError – load_from_file() raised an error that matches expected_load_error.

• Exception – Any other error raised by load_from_file().

Return type:

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

property debug: bool

The debug mode flag. Read-only attribute.

property buffer_limit: int

Maximum buffer size. Read-only attribute.

class easynetwork.serializers.base_stream.FixedSizePacketSerializer

Bases: BufferedIncrementalPacketSerializer[_T_SentDTOPacket, _T_ReceivedDTOPacket, memoryview]

A base class for stream protocols in which the packets are of a fixed size.

__init__(size: int, *, debug: bool = False, **kwargs: Any) → None

Parameters:

• size (int) – The expected data size.

• debug (bool) – If True, add information to DeserializeError via the error_info attribute.

• kwargs (Any) – Extra options given to super().__init__().

Raises:

• TypeError – Invalid integer.

• ValueError – size is negative or null.

abstractmethod serialize(packet: _T_SentDTOPacket, /) → bytes

See AbstractPacketSerializer.serialize() documentation.

Return type:

bytes

final incremental_serialize(packet: _T_SentDTOPacket, /) → Generator[bytes]

Yields the data returned by serialize().

See AbstractIncrementalPacketSerializer.incremental_serialize() documentation for details.

Raises:

• ValueError – If the returned data size is not equal to packet_size.

• Exception – Any error raised by serialize().

Return type:

Generator[bytes]

abstractmethod deserialize(data: bytes, /) → _T_ReceivedDTOPacket

See AbstractPacketSerializer.deserialize() documentation.

Return type:

_T_ReceivedDTOPacket

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

Yields until there is enough data and calls deserialize().

See AbstractIncrementalPacketSerializer.incremental_deserialize() documentation for details.

Raises:

• IncrementalDeserializeError – deserialize() raised DeserializeError.

• Exception – Any error raised by deserialize().

Return type:

Generator[None, bytes, tuple[_T_ReceivedDTOPacket, bytes]]

final create_deserializer_buffer(sizehint: int, /) → memoryview

See BufferedIncrementalPacketSerializer.buffered_incremental_deserialize() documentation for details.

Return type:

memoryview

final buffered_incremental_deserialize(buffer: memoryview, /) → Generator[int, int, tuple[_T_ReceivedDTOPacket, memoryview]]

Yields until there is enough data and calls deserialize().

See BufferedIncrementalPacketSerializer.buffered_incremental_deserialize() documentation for details.

Raises:

• IncrementalDeserializeError – deserialize() raised DeserializeError.

• Exception – Any error raised by deserialize().

Return type:

Generator[int, int, tuple[_T_ReceivedDTOPacket, memoryview]]

property packet_size: int

The expected data size. Read-only attribute.

property debug: bool

The debug mode flag. Read-only attribute.