How-to — Communication Protocols

---

The Basics

To define your communication protocol, you must instantiate one of the following protocol objects:

• DatagramProtocol: suitable for datagram oriented communication (e.g. UDP).

• StreamProtocol: suitable for stream oriented communication (e.g. TCP).

They all have one thing in common: they wrap a serializer and a converter.

You can either directly create an instance:

DatagramProtocolStreamProtocol

from __future__ import annotations

from typing import Any, TypeAlias

from easynetwork.protocol import DatagramProtocol
from easynetwork.serializers import JSONSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any

json_protocol: DatagramProtocol[SentPacket, ReceivedPacket] = DatagramProtocol(JSONSerializer())

or create a subclass:

DatagramProtocolStreamProtocol

from __future__ import annotations

from typing import Any, TypeAlias

from easynetwork.protocol import DatagramProtocol
from easynetwork.serializers import JSONSerializer

SentPacket: TypeAlias = Any
ReceivedPacket: TypeAlias = Any


class JSONDatagramProtocol(DatagramProtocol[SentPacket, ReceivedPacket]):
    def __init__(self) -> None:
        serializer = JSONSerializer()
        super().__init__(serializer)


json_protocol = JSONDatagramProtocol()

Tip

The latter is recommended. The main advantage of this model is to declaratively define the communication protocol (the name of the class being that of the protocol, the types of objects sent and received, etc.).

Another advantage is that the serializer (and converter, if any) can be configured in a single place in the project.

Usage

The protocol objects are requested by endpoint and server implementations to handle the data sent and received:

DatagramProtocolStreamProtocol

def main() -> None:
    protocol = JSONDatagramProtocol()

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

        ...

Warning

A protocol object is intended to be shared by multiple endpoints. Do not store sensitive data in these objects. You might see some magic.

Parsing Error

The protocol objects raise a BaseProtocolParseError subclass when the received data is invalid:

DatagramProtocolStreamProtocol

The raised exception is DatagramProtocolParseError.

try:
    received_packet = endpoint.recv_packet()
except DatagramProtocolParseError:
    print("The received data is invalid.")
else:
    print(f"Received {received_packet!r} from {endpoint.get_remote_address()}.")

Tip

The underlying DeserializeError instance is available with the error attribute.

The Converters

TL;DR: Why should you always have a converter in your protocol object?

Unless the serializer is already making the tea and coffee for you, in 99% of cases the data received can be anything, as long as it’s in the right format. On the other hand, the application has to comply with the format for sending data to the remote endpoint.

However, you just want to be able to manipulate your business objects without having to worry about such problems.

This is what a converter can do for you. It creates a DTO suitable for the underlying serializer and validates the received DTO to recreate the business object.

Writing A Converter

To write a converter, you must create a subclass of AbstractPacketConverter and override its convert_to_dto_packet() and create_from_dto_packet() methods.

For example:

from __future__ import annotations

import dataclasses
import uuid
from typing import Any, TypeGuard

from easynetwork.converter import AbstractPacketConverter
from easynetwork.exceptions import PacketConversionError


@dataclasses.dataclass
class Person:
    id: uuid.UUID
    name: str
    age: int
    friends: list[uuid.UUID] = dataclasses.field(default_factory=list)


class PersonToJSONConverter(AbstractPacketConverter[Person, dict[str, Any]]):
    def convert_to_dto_packet(self, person: Person, /) -> dict[str, Any]:
        return {
            "id": str(person.id),
            "name": person.name,
            "age": person.age,
            "friends": [str(friend_uuid) for friend_uuid in person.friends],
        }

    def create_from_dto_packet(self, packet: dict[str, Any], /) -> Person:
        match packet:
            case {
                "id": str(person_id),
                "name": str(name),
                "age": int(age),
                "friends": list(friends),
            } if self._is_valid_list_of_friends(friends):
                try:
                    person = Person(
                        id=uuid.UUID(person_id),
                        name=name,
                        age=age,
                        friends=[uuid.UUID(friend_id) for friend_id in friends],
                    )
                except ValueError:  # Invalid UUIDs
                    raise PacketConversionError("Invalid UUID fields") from None

            case _:
                raise PacketConversionError("Invalid packet format")

        return person

    @staticmethod
    def _is_valid_list_of_friends(friends_list: list[Any]) -> TypeGuard[list[str]]:
        return all(isinstance(friend_id, str) for friend_id in friends_list)

Warning

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

Otherwise, any other error is considered a crash.

This converter can now be used in our protocol object:

class PersonProtocol(StreamProtocol[Person, Person]):
    def __init__(self) -> None:
        serializer = JSONSerializer()
        converter = PersonToJSONConverter()

        super().__init__(serializer, converter=converter)

Note

Now this protocol is annotated to send and receive a Person object.

In the application, you can now safely handle an object with real meaning:

def main() -> None:
    protocol = PersonProtocol()

    with TCPNetworkClient(("remote_address", 12345), protocol) as endpoint:
        john_doe = Person(
            id=uuid.uuid4(),
            name="John Doe",
            age=36,
            friends=[uuid.uuid4() for _ in range(5)],
        )

        # Person object directly sent
        endpoint.send_packet(john_doe)

        try:
            # The received object should be a Person instance.
            received_person = endpoint.recv_packet()
        except StreamProtocolParseError as exc:
            match exc.error:
                case DeserializeError():
                    print("It is not even a JSON object.")
                case PacketConversionError():
                    print("It is not a valid Person in JSON object.")
        else:
            assert isinstance(received_person, Person)
            print(f"Received person: {received_person!r}")

Writing A Composite Converter

Most of the time, the packets sent and received are different (the request/response system). To deal with this, a protocol object accepts a composite converter.

To write a composite converter, there are two ways described below.

Note

Do what you think is best, there is no recommended method.

Option 1: Using StapledPacketConverter

from __future__ import annotations

from typing import Any

from easynetwork.converter import AbstractPacketConverter, StapledPacketConverter
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer


class Request:
    ...


class Response:
    ...


class RequestConverter(AbstractPacketConverter[Request, dict[str, Any]]):
    def convert_to_dto_packet(self, request: Request, /) -> dict[str, Any]:
        ...

    def create_from_dto_packet(self, request_dict: dict[str, Any], /) -> Request:
        ...


class ResponseConverter(AbstractPacketConverter[Response, dict[str, Any]]):
    def convert_to_dto_packet(self, response: Response, /) -> dict[str, Any]:
        ...

    def create_from_dto_packet(self, response_dict: dict[str, Any], /) -> Response:
        ...


serializer = JSONSerializer()
request_converter = RequestConverter()
response_converter = ResponseConverter()

client_protocol: StreamProtocol[Request, Response] = StreamProtocol(
    serializer=serializer,
    converter=StapledPacketConverter(
        sent_packet_converter=request_converter,
        received_packet_converter=response_converter,
    ),
)
server_protocol: StreamProtocol[Response, Request] = StreamProtocol(
    serializer=serializer,
    converter=StapledPacketConverter(
        sent_packet_converter=response_converter,
        received_packet_converter=request_converter,
    ),
)

Option 2: By Subclassing AbstractPacketConverterComposite

from __future__ import annotations

from typing import Any

from easynetwork.converter import AbstractPacketConverterComposite
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer


class Request:
    ...


class Response:
    ...


class ClientConverter(AbstractPacketConverterComposite[Request, Response, dict[str, Any], dict[str, Any]]):
    def convert_to_dto_packet(self, request: Request, /) -> dict[str, Any]:
        ...

    def create_from_dto_packet(self, response_dict: dict[str, Any], /) -> Response:
        ...


class ServerConverter(AbstractPacketConverterComposite[Response, Request, dict[str, Any], dict[str, Any]]):
    def convert_to_dto_packet(self, response: Response, /) -> dict[str, Any]:
        ...

    def create_from_dto_packet(self, request_dict: dict[str, Any], /) -> Request:
        ...


serializer = JSONSerializer()

client_protocol: StreamProtocol[Request, Response] = StreamProtocol(
    serializer=serializer,
    converter=ClientConverter(),
)
server_protocol: StreamProtocol[Response, Request] = StreamProtocol(
    serializer=serializer,
    converter=ServerConverter(),
)