Coverage for src / lexigram / contracts / events / messages.py: 0%

1"""Base message classes for Event Sourcing and CQRS. 

2 

3This module defines the foundational Message class and MessageMetadata 

4that all messages inherit from. These are defined in contracts to allow 

5cross-package interoperability without direct sibling dependencies. 

6 

7Also defines MessageSerializerProtocol for serializing/deserializing events. 

8""" 

9 

10from __future__ import annotations 

11 

12from dataclasses import MISSING, asdict, dataclass, field, replace 

13from datetime import UTC, datetime 

14from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, runtime_checkable 

15from uuid import UUID, uuid4 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.domain import DomainModelProtocol 

19 

20TResult = TypeVar("TResult") 

21 

22 

23@runtime_checkable 

24class MessageSerializerProtocol(Protocol): 

25 """Protocol for serializing and deserializing messages. 

26 

27 Different message formats (JSON, Avro, Protobuf) can implement 

28 this protocol to provide serialization for event stores and message buses. 

29 """ 

30 

31 def serialize(self, message: Message) -> bytes: 

32 """Serialize a message to bytes. 

33 

34 Args: 

35 message: Message to serialize. 

36 

37 Returns: 

38 Serialized message bytes. 

39 """ 

40 ... 

41 

42 def deserialize(self, data: bytes, message_type: type[Message]) -> Message: 

43 """Deserialize bytes to a message. 

44 

45 Args: 

46 data: Serialized message bytes. 

47 message_type: Type of message to deserialize to. 

48 

49 Returns: 

50 Deserialized message instance. 

51 """ 

52 ... 

53 

54 def serialize_event(self, event: DomainModelProtocol) -> bytes: 

55 """Serialize a domain event to bytes. 

56 

57 Args: 

58 event: Domain event to serialize. 

59 

60 Returns: 

61 Serialized event bytes. 

62 """ 

63 ... 

64 

65 def deserialize_event( 

66 self, data: bytes, event_type: type[DomainModelProtocol] 

67 ) -> DomainModelProtocol: 

68 """Deserialize bytes to a domain event. 

69 

70 Args: 

71 data: Serialized event bytes. 

72 event_type: Type of event to deserialize to. 

73 

74 Returns: 

75 Deserialized event instance. 

76 """ 

77 ... 

78 

79 

80@dataclass(frozen=True) 

81class MessageMetadata: 

82 """Metadata attached to messages for tracking and correlation.""" 

83 

84 correlation_id: UUID | None = None 

85 causation_id: UUID | None = None 

86 user_id: str | None = None 

87 tenant_id: str | None = None 

88 trace_id: str | None = None 

89 span_id: str | None = None 

90 custom: dict[str, Any] = field(default_factory=dict) 

91 

92 def with_correlation(self, correlation_id: UUID) -> MessageMetadata: 

93 """Create new metadata with correlation ID.""" 

94 return replace(self, correlation_id=correlation_id) 

95 

96 def with_causation(self, causation_id: UUID) -> MessageMetadata: 

97 """Create new metadata with causation ID.""" 

98 return replace(self, causation_id=causation_id) 

99 

100 def with_user(self, user_id: str) -> MessageMetadata: 

101 """Create new metadata with user ID.""" 

102 return replace(self, user_id=user_id) 

103 

104 def with_tenant(self, tenant_id: str) -> MessageMetadata: 

105 """Create new metadata with tenant ID.""" 

106 return replace(self, tenant_id=tenant_id) 

107 

108 def with_custom(self, key: str, value: Any) -> MessageMetadata: 

109 """Create new metadata with additional custom field.""" 

110 new_custom = {**self.custom, key: value} 

111 return replace(self, custom=new_custom) 

112 

113 

114@dataclass(frozen=True) 

115class Message: 

116 """Base class for all messages in the Events system.""" 

117 

118 id: UUID = field(default_factory=uuid4) 

119 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) 

120 metadata: MessageMetadata = field(default_factory=MessageMetadata) 

121 

122 def with_metadata(self, metadata: MessageMetadata) -> Message: 

123 """Create a copy of the message with new metadata.""" 

124 return replace(self, metadata=metadata) 

125 

126 def with_correlation_id(self, correlation_id: UUID) -> Message: 

127 """Create a copy with a correlation ID.""" 

128 new_metadata = self.metadata.with_correlation(correlation_id) 

129 return self.with_metadata(new_metadata) 

130 

131 def with_causation_id(self, causation_id: UUID) -> Message: 

132 """Create a copy with a causation ID.""" 

133 new_metadata = self.metadata.with_causation(causation_id) 

134 return self.with_metadata(new_metadata) 

135 

136 @property 

137 def correlation_id(self) -> UUID | None: 

138 """Get correlation ID from metadata.""" 

139 return self.metadata.correlation_id 

140 

141 @property 

142 def causation_id(self) -> UUID | None: 

143 """Get causation ID from metadata.""" 

144 return self.metadata.causation_id 

145 

146 def to_dict(self) -> dict[str, Any]: 

147 """Convert message to dictionary.""" 

148 return asdict(self) 

149 

150 @classmethod 

151 def from_dict(cls, data: dict[str, Any]) -> Message: 

152 """Create message from dictionary.""" 

153 return cls(**data) 

154 

155 

156# ``Event`` and ``DomainEventContext`` were removed in the 2026 refactor. 

157# ``DomainEvent`` in ``lexigram.contracts.domain.events`` now serves as the 

158# single canonical representation of domain events, with an optional 

159# ``actor_id`` field replacing the earlier context wrapper. 

160 

161 

162@dataclass(frozen=True) 

163class Query(Message, Generic[TResult]): 

164 """Represents a request for data.""" 

165 

166 include_deleted: bool = False 

167 cache_key: str | None = None 

168 skip_cache: bool = False 

169 

170 

171@dataclass(frozen=True) 

172class PaginatedQuery(Query[TResult]): 

173 """Query with built-in pagination support.""" 

174 

175 page: int = 1 

176 page_size: int = 20 

177 sort_by: str | None = None 

178 sort_desc: bool = False 

179 

180 

181@dataclass(init=False, frozen=True) 

182class Command(Message, Generic[TResult]): 

183 """Represents an intent to change state. 

184 

185 A frozen dataclass that accepts all declared fields as keyword arguments, 

186 plus any extra kwargs (stored as object attributes for forward-compatibility). 

187 We intentionally disable dataclass-generated ``__init__`` and supply our 

188 own so that subclasses can be instantiated with their specific fields without 

189 receiving ``TypeError`` for unknown extras. 

190 

191 Note: This class intentionally does *not* inherit from ``DomainModel`` 

192 (which lives in ``lexigram``, not ``lexigram-contracts``) to avoid a 

193 circular dependency. All dataclass bookkeeping is done directly here. 

194 """ 

195 

196 def __init__(self, *args: Any, **kwargs: Any) -> None: 

197 dc_fields = getattr(self.__class__, "__dataclass_fields__", {}) 

198 for fname, f in dc_fields.items(): 

199 if fname in kwargs: 

200 object.__setattr__(self, fname, kwargs.pop(fname)) 

201 elif f.default is not MISSING: 

202 object.__setattr__(self, fname, f.default) 

203 elif f.default_factory is not MISSING: 

204 object.__setattr__(self, fname, f.default_factory()) 

205 # else: required field not provided — TypeError raised naturally 

206 # Store any remaining extra kwargs as attributes (forward compatibility) 

207 for k, v in kwargs.items(): 

208 object.__setattr__(self, k, v) 

209 

210 

211@dataclass(frozen=True) 

212class IdempotentCommand(Command[TResult]): 

213 """Represents a command that can be safely retried.""" 

214 

215 idempotency_key: str | None = None 

216 

217 

218__all__ = [ 

219 "Command", 

220 "IdempotentCommand", 

221 "Message", 

222 "MessageMetadata", 

223 "MessageSerializerProtocol", 

224 "PaginatedQuery", 

225 "Query", 

226]