Skip to content

Dict Message Codec

_dict_message_codec

Dict-based message codec that serializes messages to dict[str, object].

DictMessageCodec

Bases: MessageCodec[M, dict[str, object]]

Codec that serializes messages to dict[str, object].

Uses the message's own value property for the payload and
message.metadata.value for the metadata section. Reconstruction
goes through the from_payload_fields classmethod that every
concrete Message subclass provides.

Example 
class StubCommand[T](Message[T]):
    # Inline stub for the example.
    pass


class CreateOrder(StubCommand[dict[str, object]]):
    def __init__(self, customer_id: str) -> None:
        super().__init__()
        self.customer_id = customer_id

    @property
    def _payload(self) -> dict[str, object]:
        return {"customer_id": self.customer_id}

    @classmethod
    def from_payload_fields(
        cls, payload: dict[str, object], metadata: MessageMetadata
    ) -> CreateOrder:
        return cls(customer_id=str(payload["customer_id"]))

    @property
    def value(self) -> dict[str, object]:
        return self._payload


codec = DictMessageCodec[CreateOrder]()
original = CreateOrder(customer_id="cust-42")
encoded = codec.encode(original)
decoded = codec.decode(encoded, CreateOrder)
assert decoded.customer_id == original.customer_id
Source code in src/forging_blocks/infrastructure/serialization/_dict_message_codec.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class DictMessageCodec[M: Message[dict[str, object]]](MessageCodec[M, dict[str, object]]):
    """Codec that serializes messages to ``dict[str, object]``.

    Uses the message's own ``value`` property for the payload and
    ``message.metadata.value`` for the metadata section.  Reconstruction
    goes through the ``from_payload_fields`` classmethod that every
    concrete ``Message`` subclass provides.

    Example:
        ```python
        class StubCommand[T](Message[T]):
            # Inline stub for the example.
            pass


        class CreateOrder(StubCommand[dict[str, object]]):
            def __init__(self, customer_id: str) -> None:
                super().__init__()
                self.customer_id = customer_id

            @property
            def _payload(self) -> dict[str, object]:
                return {"customer_id": self.customer_id}

            @classmethod
            def from_payload_fields(
                cls, payload: dict[str, object], metadata: MessageMetadata
            ) -> CreateOrder:
                return cls(customer_id=str(payload["customer_id"]))

            @property
            def value(self) -> dict[str, object]:
                return self._payload


        codec = DictMessageCodec[CreateOrder]()
        original = CreateOrder(customer_id="cust-42")
        encoded = codec.encode(original)
        decoded = codec.decode(encoded, CreateOrder)
        assert decoded.customer_id == original.customer_id
        ```

    """

    def encode(self, message: M) -> dict[str, object]:
        """Encode *message* to a dictionary with ``metadata`` and ``payload`` keys."""
        return {
            "metadata": message.metadata.value,
            "payload": message.value,
        }

    def decode(self, data: dict[str, object], message_type: type[M]) -> M:
        """Decode *data* back into a message of *message_type*."""
        raw_metadata = cast(dict[str, object], data["metadata"])
        payload = cast(dict[str, object], data.get("payload", {}))

        metadata = MessageMetadata(
            message_type=str(raw_metadata.get("message_type", message_type.__name__)),
            message_id=UUID(
                str(_get_required(raw_metadata, "message_id")),
            ),
            created_at=datetime.fromisoformat(
                str(_get_required(raw_metadata, "created_at")),
            ),
            causation_id=UUID(
                str(_get_required(raw_metadata, "causation_id")),
            ),
            correlation_id=UUID(
                str(_get_required(raw_metadata, "correlation_id")),
            ),
        )

        return message_type.from_payload_fields(payload, metadata)

encode(message: M) -> dict[str, object]

Encode message to a dictionary with metadata and payload keys.

Source code in src/forging_blocks/infrastructure/serialization/_dict_message_codec.py 
64
65
66
67
68
69
def encode(self, message: M) -> dict[str, object]:
    """Encode *message* to a dictionary with ``metadata`` and ``payload`` keys."""
    return {
        "metadata": message.metadata.value,
        "payload": message.value,
    }

decode(data: dict[str, object], message_type: type[M]) -> M

Decode data back into a message of message_type.

Source code in src/forging_blocks/infrastructure/serialization/_dict_message_codec.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def decode(self, data: dict[str, object], message_type: type[M]) -> M:
    """Decode *data* back into a message of *message_type*."""
    raw_metadata = cast(dict[str, object], data["metadata"])
    payload = cast(dict[str, object], data.get("payload", {}))

    metadata = MessageMetadata(
        message_type=str(raw_metadata.get("message_type", message_type.__name__)),
        message_id=UUID(
            str(_get_required(raw_metadata, "message_id")),
        ),
        created_at=datetime.fromisoformat(
            str(_get_required(raw_metadata, "created_at")),
        ),
        causation_id=UUID(
            str(_get_required(raw_metadata, "causation_id")),
        ),
        correlation_id=UUID(
            str(_get_required(raw_metadata, "correlation_id")),
        ),
    )

    return message_type.from_payload_fields(payload, metadata)