Skip to content

Metadata

_metadata

MessageMetadata value object for messaging patterns.

MessageMetadata

Bases: ValueObject[dict[str, object]]

Metadata associated with foundational messages.

Contains technical-level information about messages such as:
- Unique message identifier
- When the message was created
- correlation_id is used to trace related messages across systems
and link messages that belong to the same business process.
- causation_id is used to identify the immediate predecessor message that caused
this message.

This separation allows messages to focus on foundational data while keeping
infrastructure handling concerns in metadata without needing to understand anything about
infrastructure rules.

Example 
metadata = MessageMetadata(message_type="OrderCreated")

# Or with custom values
custom_metadata = MessageMetadata(
    message_type="UserCreated",
    message_id=UUID("123e4567-e89b-12d3-a456-426614174000"),
    created_at=datetime(2025, 6, 11, 19, 36, 6, tzinfo=timezone.utc),
)
Source code in src/forging_blocks/domain/messages/message/_metadata.py 
  9
 10
 11
 12
 13
 14
 15
 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
class MessageMetadata(ValueObject[dict[str, object]]):
    """Metadata associated with foundational messages.

    Contains technical-level information about messages such as:
    - Unique message identifier
    - When the message was created
    - correlation_id is used to trace related messages across systems
      and link messages that belong to the same business process.
    - causation_id is used to identify the immediate predecessor message that caused
      this message.

    This separation allows messages to focus on foundational data while keeping
    infrastructure handling concerns in metadata without needing to understand anything about
    infrastructure rules.

    Example:
        ```python
        metadata = MessageMetadata(message_type="OrderCreated")

        # Or with custom values
        custom_metadata = MessageMetadata(
            message_type="UserCreated",
            message_id=UUID("123e4567-e89b-12d3-a456-426614174000"),
            created_at=datetime(2025, 6, 11, 19, 36, 6, tzinfo=timezone.utc),
        )
        ```

    """

    __slots__ = (
        "_message_type",
        "_message_id",
        "_created_at",
        "_correlation_id",
        "_causation_id",
    )

    def __init__(
        self,
        message_type: str,
        message_id: UUID | None = None,
        created_at: datetime | None = None,
        correlation_id: UUID | None = None,
        causation_id: UUID | None = None,
    ) -> None:
        """Initialize message metadata.

        Args:
            message_type: The type/name of the message.
            message_id: Unique identifier for the message. If None, generates a
                new UUID.
            created_at: When the message was created. If None, uses current UTC time.
            correlation_id: Identifier to correlate related messages. If None,
                generates a new UUID.
            causation_id: Identifier of the message that caused this one. If None,
                generates a new UUID.

        """
        super().__init__()
        self._message_type = message_type
        self._message_id = message_id if message_id is not None else uuid7()
        self._created_at = created_at if created_at is not None else datetime.now(timezone.utc)
        self._correlation_id = correlation_id if correlation_id is not None else uuid7()
        self._causation_id = causation_id if causation_id is not None else uuid7()

    @property
    def message_id(self) -> UUID:
        """Get the unique identifier for this message.

        Returns:
            The unique message identifier.

        """
        return self._message_id

    @property
    def causation_id(self) -> UUID:
        """Get the causation ID for this message.

        Returns:
            The causation identifier.

        """
        return self._causation_id

    @property
    def created_at(self) -> datetime:
        """Get the timestamp when this message was created.

        Returns:
            The timestamp (preserved as given when provided, or the
            current UTC time when defaulted).

        """
        return self._created_at

    @property
    def correlation_id(self) -> UUID:
        """Get the correlation ID for this message.

        Returns:
            The correlation identifier.

        """
        return self._correlation_id

    @property
    def message_type(self) -> str:
        """Get the type of this message.

        Returns:
            The message type name.

        """
        return self._message_type

    @property
    def value(self) -> dict[str, object]:
        """Get the raw dictionary representation of the metadata."""
        return {
            "created_at": self._created_at.isoformat(),
            "correlation_id": str(self._correlation_id),
            "causation_id": str(self._causation_id),
            "message_id": str(self._message_id),
            "message_type": self._message_type,
        }

message_id: UUID property

Get the unique identifier for this message.

Returns:

Type Description
UUID

The unique message identifier.

causation_id: UUID property

Get the causation ID for this message.

Returns:

Type Description
UUID

The causation identifier.

created_at: datetime property

Get the timestamp when this message was created.

Returns:

Type Description
datetime

The timestamp (preserved as given when provided, or the
datetime

current UTC time when defaulted).

correlation_id: UUID property

Get the correlation ID for this message.

Returns:

Type Description
UUID

The correlation identifier.

message_type: str property

Get the type of this message.

Returns:

Type Description
str

The message type name.

value: dict[str, object] property

Get the raw dictionary representation of the metadata.

__init__(message_type: str, message_id: UUID | None = None, created_at: datetime | None = None, correlation_id: UUID | None = None, causation_id: UUID | None = None) -> None

Initialize message metadata.

Parameters:

Name Type Description Default
message_type str

The type/name of the message.

 required
message_id UUID | None

Unique identifier for the message. If None, generates a
new UUID.

 None
created_at datetime | None

When the message was created. If None, uses current UTC time.

 None
correlation_id UUID | None

Identifier to correlate related messages. If None,
generates a new UUID.

 None
causation_id UUID | None

Identifier of the message that caused this one. If None,
generates a new UUID.

 None
Source code in src/forging_blocks/domain/messages/message/_metadata.py 
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
def __init__(
    self,
    message_type: str,
    message_id: UUID | None = None,
    created_at: datetime | None = None,
    correlation_id: UUID | None = None,
    causation_id: UUID | None = None,
) -> None:
    """Initialize message metadata.

    Args:
        message_type: The type/name of the message.
        message_id: Unique identifier for the message. If None, generates a
            new UUID.
        created_at: When the message was created. If None, uses current UTC time.
        correlation_id: Identifier to correlate related messages. If None,
            generates a new UUID.
        causation_id: Identifier of the message that caused this one. If None,
            generates a new UUID.

    """
    super().__init__()
    self._message_type = message_type
    self._message_id = message_id if message_id is not None else uuid7()
    self._created_at = created_at if created_at is not None else datetime.now(timezone.utc)
    self._correlation_id = correlation_id if correlation_id is not None else uuid7()
    self._causation_id = causation_id if causation_id is not None else uuid7()