How-to — Serializers

---

The Basics

Where Are They Used?

A serializer is used by a protocol object:

• DatagramProtocol: accepts any serializer object.

• StreamProtocol: accepts incremental serializer objects only.

Which Ones Are Already Available?

Several serializers are provided in the easynetwork.serializers module. Do not hesitate to use them.

If nothing fits your needs, you can implement your own serializer.

Writing A One-Shot Serializer

One-shot serializers are the easiest piece of code to write. They can be used directly on DatagramProtocol instances, and you can use a serializer wrapper to use the StreamProtocol class.

See also

easynetwork.serializers.wrapper module

The full list of available wrappers for serializers.

To write a one-shot serializer, you must create a subclass of AbstractPacketSerializer and override its serialize() and deserialize() methods.

A naive implementation of JSONSerializer should look something like this:

from __future__ import annotations

import json
from typing import Any, TypeAlias

from easynetwork.serializers.abc import AbstractPacketSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any


class MyJSONSerializer(AbstractPacketSerializer[SentPacket, ReceivedPacket]):
    def serialize(self, packet: SentPacket) -> bytes:
        document = json.dumps(packet)
        return document.encode()

    def deserialize(self, data: bytes) -> ReceivedPacket:
        document = data.decode()
        return json.loads(document)

Parsing Error

The deserialize() function must raise a DeserializeError to indicate that a parsing error was “expected” so that the received data is considered invalid.

from __future__ import annotations

import json
from typing import Any, TypeAlias

from easynetwork.exceptions import DeserializeError
from easynetwork.serializers.abc import AbstractPacketSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any


class MyJSONSerializer(AbstractPacketSerializer[SentPacket, ReceivedPacket]):
    def serialize(self, packet: SentPacket) -> bytes:
        document = json.dumps(packet)
        return document.encode()

    def deserialize(self, data: bytes) -> ReceivedPacket:
        try:
            document = data.decode()
            return json.loads(document)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise DeserializeError("JSON decode error") from exc

Warning

Otherwise, any other error is considered a serializer crash.

The Use Of self

A serializer is intended to be shared by multiple protocols (and protocols are intended to be shared by multiple endpoints).

Therefore, the object should only store additional configuration used for serialization/deserialization.

For example:

from __future__ import annotations

import json
from typing import Any, TypeAlias

from easynetwork.exceptions import DeserializeError
from easynetwork.serializers.abc import AbstractPacketSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any


class MyJSONSerializer(AbstractPacketSerializer[SentPacket, ReceivedPacket]):
    def __init__(self, *, ensure_ascii: bool = True) -> None:
        self._ensure_ascii: bool = ensure_ascii

        self._encoding: str
        if self._ensure_ascii:
            self._encoding = "ascii"
        else:
            self._encoding = "utf-8"

    def serialize(self, packet: SentPacket) -> bytes:
        document = json.dumps(packet, ensure_ascii=self._ensure_ascii)
        return document.encode(self._encoding)

    def deserialize(self, data: bytes) -> ReceivedPacket:
        try:
            document = data.decode(self._encoding)
            return json.loads(document)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise DeserializeError("JSON decode error") from exc

Warning

Do not store per-serialization data. You might see some magic.

Danger

Seriously, don’t do that.

Using A One-Shot Serializer

You must pass the serializer to the appropriate protocol object that is expected by the endpoint class:

DatagramProtocolStreamProtocol

def main() -> None:
    serializer = MyJSONSerializer()
    protocol = DatagramProtocol(serializer)

    with UDPNetworkClient(("remote_address", 12345), protocol) as endpoint:
        endpoint.send_packet({"data": 42})

        ...

Note

Using a serializer wrapper means that the transferred data can be completely different from the original output.

If this is important to you, don’t choose one of them lightly.

Writing An Incremental Serializer

Incremental serializers are a bit trickier to implement. They can be used directly on both StreamProtocol and DatagramProtocol instances.

To write an incremental serializer, you must create a subclass of AbstractIncrementalPacketSerializer and override its incremental_serialize() and incremental_deserialize() methods. The serialize() and deserialize() methods have a default implementation that uses the incremental serialization methods.

Option 1: Using Common Patterns

Chances are that the communication protocol uses a simple principle to determine the end of a packet. The most common cases are:

• All your packet frames use a precise byte sequence (most likely a newline).

• Each packet has a fixed size.

In these cases you can use the base classes in easynetwork.serializers.base_stream.

Let’s say that for the incremental part, we consider each line received to be a JSON object, separated by \r\n:

from __future__ import annotations

import json
from typing import Any, TypeAlias

from easynetwork.exceptions import DeserializeError
from easynetwork.serializers.base_stream import AutoSeparatedPacketSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any


class MyJSONSerializer(AutoSeparatedPacketSerializer[SentPacket, ReceivedPacket]):
    def __init__(self, *, ensure_ascii: bool = True) -> None:
        super().__init__(separator=b"\r\n")

        self._ensure_ascii: bool = ensure_ascii

        self._encoding: str
        if self._ensure_ascii:
            self._encoding = "ascii"
        else:
            self._encoding = "utf-8"

    def serialize(self, packet: SentPacket) -> bytes:
        document = json.dumps(packet, ensure_ascii=self._ensure_ascii)
        return document.encode(self._encoding)

    def deserialize(self, data: bytes) -> ReceivedPacket:
        try:
            document = data.decode(self._encoding)
            return json.loads(document)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise DeserializeError("JSON decode error") from exc

AutoSeparatedPacketSerializer adds the following behaviors:

• incremental_serialize() will append \r\n to the end of the serialize() output.

• incremental_deserialize() waits until \r\n is found in the input, removes the separator, and calls deserialize() on it.

Tip

Take a look at other available base classes in easynetwork.serializers before rewriting something that already exists.

Option 2: From Scratch

Let’s see how we can get by without using the AutoSeparatedPacketSerializer:

from __future__ import annotations

import json
from collections.abc import Generator
from typing import Any, TypeAlias

from easynetwork.exceptions import DeserializeError, IncrementalDeserializeError
from easynetwork.serializers.abc import AbstractIncrementalPacketSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any


class MyJSONSerializer(AbstractIncrementalPacketSerializer[SentPacket, ReceivedPacket]):
    def __init__(self, *, ensure_ascii: bool = True) -> None:
        self._ensure_ascii: bool = ensure_ascii

        self._encoding: str
        if self._ensure_ascii:
            self._encoding = "ascii"
        else:
            self._encoding = "utf-8"

    def _dump(self, packet: SentPacket) -> bytes:
        document = json.dumps(packet, ensure_ascii=self._ensure_ascii)
        return document.encode(self._encoding)

    def _load(self, data: bytes) -> ReceivedPacket:
        document = data.decode(self._encoding)
        return json.loads(document)

    def serialize(self, packet: SentPacket) -> bytes:
        return self._dump(packet)

    def deserialize(self, data: bytes) -> ReceivedPacket:
        try:
            return self._load(data)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise DeserializeError("JSON decode error") from exc

    def incremental_serialize(self, packet: SentPacket) -> Generator[bytes, None, None]:
        yield self._dump(packet) + b"\r\n"

    def incremental_deserialize(self) -> Generator[None, bytes, tuple[ReceivedPacket, bytes]]:
        data = yield
        newline = b"\r\n"
        while (index := data.find(newline)) < 0:
            data += yield

        remainder = data[index + len(newline) :]
        data = data[:index]

        try:
            document = self._load(data)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise IncrementalDeserializeError("JSON decode error", remainder) from exc

        return document, remainder

This adds a lot of code! Let’s take a closer look at the implementation.

Code Mutualization

To avoid duplication of code between the one-shot part and the incremental part, the serialization/deserialization part of the code goes to a private method.

def _dump(self, packet: SentPacket) -> bytes:
    document = json.dumps(packet, ensure_ascii=self._ensure_ascii)
    return document.encode(self._encoding)

def _load(self, data: bytes) -> ReceivedPacket:
    document = data.decode(self._encoding)
    return json.loads(document)

And now serialize() and deserialize() use them instead:

def serialize(self, packet: SentPacket) -> bytes:
    return self._dump(packet)

def deserialize(self, data: bytes) -> ReceivedPacket:
    try:
        return self._load(data)
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise DeserializeError("JSON decode error") from exc

The Purpose Of incremental_serialize()

incremental_serialize() must be a generator function (or at least return a generator iterator) that yields all the parts of the serialized packet. It must also add any useful metadata to help incremental_deserialize() find the end of the packet.

def incremental_serialize(self, packet: SentPacket) -> Generator[bytes, None, None]:
    yield self._dump(packet) + b"\r\n"

Most of the time, you will have a single yield. The goal is: each yield must send as many bytes as possible without copying or concatenating.

Tip

There may be exceptions, like this example. (Your RAM will not cry because you added 2 bytes to a byte sequence of almost 100 bytes. The question may be asked if the byte sequence is ending up to 4 GB.)

It is up to you to find the balance between RAM explosion and performance degradation.

Note

The endpoint implementation can decide to concatenate all the pieces and do one big send. However, it may be more attractive to do something else with the returned bytes. incremental_serialize() is here to give endpoints this freedom.

The Purpose Of incremental_deserialize()

incremental_deserialize() must be a generator function (or at least return a generator iterator) that yields None until all the data parts of the packet have been retrieved and parsed.

This generator must return a pair of (packet, remainder) where packet is the deserialized packet and remainder is any superfluous trailing bytes that was useless.

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

def incremental_deserialize(self) -> Generator[None, bytes, tuple[ReceivedPacket, bytes]]:
    data = yield
    newline = b"\r\n"
    while (index := data.find(newline)) < 0:
        data += yield

    remainder = data[index + len(newline) :]
    data = data[:index]

    try:
        document = self._load(data)
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise IncrementalDeserializeError("JSON decode error", remainder) from exc

    return document, remainder

Note

Even if we could create 5 more JSON packets from remainder, incremental_deserialize() must always deserialize the first one available and return the rest as is.

This allows the endpoint implementation to deserialize only the needed packet. The rest is reused when the application wants an other packet.

See also

Serializer implementation tools

Regroups helpers for (incremental) serializer implementations.

PEP 255 — Simple Generators

The proposal for adding generators and the yield statement to Python.

PEP 342 — Coroutines via Enhanced Generators

The proposal to enhance the API and syntax of generators, making them usable as simple coroutines.

PEP 380 — Syntax for Delegating to a Subgenerator

The proposal to introduce the yield_from syntax, making delegation to subgenerators easy.

Parsing Error

The incremental_deserialize() function must raise an IncrementalDeserializeError to indicate that a parsing error was “expected” so that the received data is considered invalid.

def incremental_deserialize(self) -> Generator[None, bytes, tuple[ReceivedPacket, bytes]]:
    data = yield
    newline = b"\r\n"
    while (index := data.find(newline)) < 0:
        data += yield

    remainder = data[index + len(newline) :]
    data = data[:index]

    try:
        document = self._load(data)
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise IncrementalDeserializeError("JSON decode error", remainder) from exc

    return document, remainder

Warning

Otherwise, any other error is considered a serializer crash.

If DeserializeError is raised instead, this is converted to a RuntimeError.

Note

IncrementalDeserializeError needs the possible valid remainder, that is not the root cause of the error. In the example, even if data is an invalid JSON object, all bytes after the \r\n token (in remainder) are not lost.