Skip to content

Message

_message

Base Message class for messaging patterns.

Message

Bases: ABC

Base class for all foundation messages.

Messages represent intent or facts in the application. This is the
base class for Commands (something to do), Events (something that
happened), and Queries (something to obtain).

Messages are immutable and each instance is unique — equality and
hash are determined solely by the message_id carried in
MessageMetadata, enforced via auto_hash and
auto_eq with fields=["message_id"].

This class should not be used directly. Import Event or
Command instead.

Example 
from forging_blocks.domain.messages.command import Command


class PlaceOrder(Command[str]):
    def __init__(self, description: str) -> None:
        super().__init__()
        self.description = description

    @property
    def _payload(self) -> str:
        return self.description

    @classmethod
    def from_payload_fields(
        cls, payload: str, metadata: MessageMetadata | None = None
    ) -> PlaceOrder:
        return cls(payload)

    @property
    def value(self) -> str:
        return self.description


cmd = PlaceOrder("Buy groceries")
print(cmd.message_id)  # unique identifier
print(cmd.description)  # "Buy groceries"
Source code in src/forging_blocks/domain/messages/message/_message.py 
 16
 17
 18
 19
 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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class Message[MessageRawType](ABC):
    """Base class for all foundation messages.

    Messages represent intent or facts in the application.  This is the
    base class for Commands (something to do), Events (something that
    happened), and Queries (something to obtain).

    Messages are immutable and each instance is unique — equality and
    hash are determined solely by the ``message_id`` carried in
    `MessageMetadata`, enforced via `auto_hash` and
    `auto_eq` with ``fields=["message_id"]``.

    This class should not be used directly.  Import `Event` or
    `Command` instead.

    Example:
        ```python
        from forging_blocks.domain.messages.command import Command


        class PlaceOrder(Command[str]):
            def __init__(self, description: str) -> None:
                super().__init__()
                self.description = description

            @property
            def _payload(self) -> str:
                return self.description

            @classmethod
            def from_payload_fields(
                cls, payload: str, metadata: MessageMetadata | None = None
            ) -> PlaceOrder:
                return cls(payload)

            @property
            def value(self) -> str:
                return self.description


        cmd = PlaceOrder("Buy groceries")
        print(cmd.message_id)  # unique identifier
        print(cmd.description)  # "Buy groceries"
        ```
    """

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Automatically apply ``auto_hash``, ``auto_eq``, and ``auto_freeze``
        to concrete subclasses.

        ``auto_hash`` and ``auto_eq`` are applied unconditionally (before the
        abstract-method check) so they take effect even when a decorator like
        ``@message_dataclass`` patches ``__abstractmethods__`` later.
        ``auto_hash`` and ``auto_eq`` use ``fields=["message_id"]`` so that
        message identity (equality and hashing) is driven solely by the
        unique message identifier, not by payload fields.
        """
        super().__init_subclass__(**kwargs)
        auto_hash(cls, fields=["message_id"])
        auto_eq(cls, fields=["message_id"])
        if not inspect.isabstract(cls):
            auto_freeze(cls)

    def __init__(self, metadata: MessageMetadata | None = None) -> None:
        """Initialize the message with metadata.

        Args:
            metadata: Message metadata. If None, creates new metadata with
                generated ID and current timestamp.

        """
        super().__init__()
        effective_type = type(self).__name__
        self._metadata = metadata or MessageMetadata(message_type=effective_type)

    @property
    def metadata(self) -> MessageMetadata:
        """Get the message metadata.

        Returns:
            The message metadata containing ID, timestamp, etc.

        """
        return self._metadata

    @property
    def message_id(self) -> UUID:
        """Convenience property to get the message ID.

        Returns:
            The unique message identifier.

        """
        return self._metadata.message_id

    @property
    def created_at(self) -> datetime:
        """Convenience property to get when the message was created.

        Returns:
            When the message was created.

        """
        return self._metadata.created_at

    @property
    @abstractmethod
    def _payload(self) -> MessageRawType:
        """Get the data carried by this message.

        Subclasses must implement this property to provide their specific message
        data. This makes the Message class truly abstract.

        Returns:
            The message payload.

        """

    @classmethod
    @abstractmethod
    def from_payload_fields(
        cls,
        data: MessageRawType,
        metadata: MessageMetadata,
    ) -> Self:
        """Reconstruct a message instance from payload fields and metadata.

        Abstract classmethod that subclasses must implement.  The
        ``@message_dataclass`` decorator provides a concrete implementation
        automatically; manual subclasses that need codec support must override
        this method themselves.

        Returns:
            A new message instance reconstructed from the given payload
            fields and metadata.

        """

    @property
    @abstractmethod
    def value(self) -> MessageRawType:
        """Return the raw message payload as a single value."""

metadata: MessageMetadata property

Get the message metadata.

Returns:

Type Description
MessageMetadata

The message metadata containing ID, timestamp, etc.

message_id: UUID property

Convenience property to get the message ID.

Returns:

Type Description
UUID

The unique message identifier.

created_at: datetime property

Convenience property to get when the message was created.

Returns:

Type Description
datetime

When the message was created.

value: MessageRawType abstractmethod property

Return the raw message payload as a single value.

__init_subclass__(**kwargs: Any) -> None

Automatically apply auto_hash, auto_eq, and auto_freeze
to concrete subclasses.

auto_hash and auto_eq are applied unconditionally (before the
abstract-method check) so they take effect even when a decorator like
@message_dataclass patches __abstractmethods__ later.
auto_hash and auto_eq use fields=["message_id"] so that
message identity (equality and hashing) is driven solely by the
unique message identifier, not by payload fields.

Source code in src/forging_blocks/domain/messages/message/_message.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init_subclass__(cls, **kwargs: Any) -> None:
    """Automatically apply ``auto_hash``, ``auto_eq``, and ``auto_freeze``
    to concrete subclasses.

    ``auto_hash`` and ``auto_eq`` are applied unconditionally (before the
    abstract-method check) so they take effect even when a decorator like
    ``@message_dataclass`` patches ``__abstractmethods__`` later.
    ``auto_hash`` and ``auto_eq`` use ``fields=["message_id"]`` so that
    message identity (equality and hashing) is driven solely by the
    unique message identifier, not by payload fields.
    """
    super().__init_subclass__(**kwargs)
    auto_hash(cls, fields=["message_id"])
    auto_eq(cls, fields=["message_id"])
    if not inspect.isabstract(cls):
        auto_freeze(cls)

__init__(metadata: MessageMetadata | None = None) -> None

Initialize the message with metadata.

Parameters:

Name Type Description Default
metadata MessageMetadata | None

Message metadata. If None, creates new metadata with
generated ID and current timestamp.

 None
Source code in src/forging_blocks/domain/messages/message/_message.py 
79
80
81
82
83
84
85
86
87
88
89
def __init__(self, metadata: MessageMetadata | None = None) -> None:
    """Initialize the message with metadata.

    Args:
        metadata: Message metadata. If None, creates new metadata with
            generated ID and current timestamp.

    """
    super().__init__()
    effective_type = type(self).__name__
    self._metadata = metadata or MessageMetadata(message_type=effective_type)

from_payload_fields(data: MessageRawType, metadata: MessageMetadata) -> Self abstractmethod classmethod

Reconstruct a message instance from payload fields and metadata.

Abstract classmethod that subclasses must implement. The
@message_dataclass decorator provides a concrete implementation
automatically; manual subclasses that need codec support must override
this method themselves.

Returns:

Type Description
Self

A new message instance reconstructed from the given payload
Self

fields and metadata.
Source code in src/forging_blocks/domain/messages/message/_message.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@classmethod
@abstractmethod
def from_payload_fields(
    cls,
    data: MessageRawType,
    metadata: MessageMetadata,
) -> Self:
    """Reconstruct a message instance from payload fields and metadata.

    Abstract classmethod that subclasses must implement.  The
    ``@message_dataclass`` decorator provides a concrete implementation
    automatically; manual subclasses that need codec support must override
    this method themselves.

    Returns:
        A new message instance reconstructed from the given payload
        fields and metadata.

    """