Message Codec

_message_codec

Message codec ABC for encoding/decoding foundation messages.

MessageCodec

Bases: ABC

Abstract codec for encoding and decoding messages.

A codec is a bidirectional transformation between a message
instance and a raw representation. Subclasses implement the
serialization format (dict, bytes, wire protocol, etc.).

Class Type Parameters:

Name	Bound or Constraints	Description	Default
M		The message type this codec handles.	required
Raw		The raw representation produced/consumed (e.g., dict, bytes, str).	required

Example

class Notification:
    def __init__(self, topic: str, body: str) -> None:
        self.topic = topic
        self.body = body


class JsonMessageCodec:
    def encode(self, message): ...
    def decode(self, data, message_type): ...


codec = JsonMessageCodec()
raw = codec.encode(Notification(topic="orders", body="Shipped"))
msg = codec.decode(raw, Notification)

Source code in src/forging_blocks/infrastructure/serialization/_message_codec.py

class MessageCodec[M, Raw](ABC):
    """Abstract codec for encoding and decoding messages.

    A *codec* is a bidirectional transformation between a message
    instance and a raw representation.  Subclasses implement the
    serialization format (dict, bytes, wire protocol, etc.).

    Type Parameters:
        M: The message type this codec handles.
        Raw: The raw representation produced/consumed (e.g., ``dict``,
            ``bytes``, ``str``).

    Example:
        ```python
        class Notification:
            def __init__(self, topic: str, body: str) -> None:
                self.topic = topic
                self.body = body


        class JsonMessageCodec:
            def encode(self, message): ...
            def decode(self, data, message_type): ...


        codec = JsonMessageCodec()
        raw = codec.encode(Notification(topic="orders", body="Shipped"))
        msg = codec.decode(raw, Notification)
        ```
    """

    @abstractmethod
    def encode(self, message: M) -> Raw:
        """Encode *message* to its raw representation.

        Args:
            message: The message instance to encode.

        Returns:
            The raw representation of the message.
        """
        ...

    @abstractmethod
    def decode(self, data: Raw, message_type: type[M]) -> M:
        """Decode *data* back into a message of *message_type*.

        Args:
            data: The raw data to decode.
            message_type: The target message class (used to dispatch).

        Returns:
            A new message instance of type *message_type*.
        """
        ...

encode(message: M) -> Raw abstractmethod

Encode message to its raw representation.

Parameters:

Name	Type	Description	Default
message	M	The message instance to encode.	required

Returns:

Type	Description
Raw	The raw representation of the message.

Source code in src/forging_blocks/infrastructure/serialization/_message_codec.py

@abstractmethod
def encode(self, message: M) -> Raw:
    """Encode *message* to its raw representation.

    Args:
        message: The message instance to encode.

    Returns:
        The raw representation of the message.
    """
    ...

decode(data: Raw, message_type: type[M]) -> M abstractmethod

Decode data back into a message of message_type.

Parameters:

Name	Type	Description	Default
data	Raw	The raw data to decode.	required
message_type	type[M]	The target message class (used to dispatch).	required

Returns:

Type	Description
M	A new message instance of type message_type.

Source code in src/forging_blocks/infrastructure/serialization/_message_codec.py

@abstractmethod
def decode(self, data: Raw, message_type: type[M]) -> M:
    """Decode *data* back into a message of *message_type*.

    Args:
        data: The raw data to decode.
        message_type: The target message class (used to dispatch).

    Returns:
        A new message instance of type *message_type*.
    """
    ...