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

1"""Event sourcing protocol class definitions (CQRS/ES). 

2 

3Protocols for event buses, command buses, event stores, 

4and related patterns. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable 

10 

11if TYPE_CHECKING: 

12 from datetime import datetime 

13 from uuid import UUID 

14 

15 from lexigram.contracts.core.result import Result 

16 from lexigram.contracts.exceptions.events import EventError 

17 

18TAggregate_co = TypeVar("TAggregate_co", covariant=True) 

19 

20# --- Domain Events --- 

21 

22 

23@runtime_checkable 

24class DomainEventPublisherProtocol(Protocol): 

25 """Protocol for publishing domain events.""" 

26 

27 async def publish(self, event: Any) -> None: 

28 """Publish a domain event. 

29 

30 Args: 

31 event: DomainEvent instance. 

32 """ 

33 ... 

34 

35 

36@runtime_checkable 

37class EventHandlerProtocol(Protocol): 

38 """Protocol for event handlers.""" 

39 

40 async def handle(self, event: Any) -> Result[None, EventError]: 

41 """Handle an event. 

42 

43 Args: 

44 event: DomainEvent to handle. 

45 

46 Returns: 

47 ``Ok(None)`` on success, ``Err(EventError)`` if handling fails 

48 in an expected, recoverable way. 

49 """ 

50 ... 

51 

52 

53@runtime_checkable 

54class MultiEventHandlerProtocol(Protocol): 

55 """Protocol for handlers that handle multiple event types.""" 

56 

57 def handles(self) -> list[type]: 

58 """Get list of event types handled. 

59 

60 Returns: 

61 List of event classes. 

62 """ 

63 ... 

64 

65 async def handle(self, event: Any) -> Result[None, EventError]: 

66 """Handle an event. 

67 

68 Args: 

69 event: DomainEvent to handle. 

70 

71 Returns: 

72 ``Ok(None)`` on success, ``Err(EventError)`` if handling fails 

73 in an expected, recoverable way. 

74 """ 

75 ... 

76 

77 

78@runtime_checkable 

79class EventBusProtocol(Protocol): 

80 """Protocol for event bus implementations. 

81 

82 The event bus manages event publication and subscription. 

83 

84 Example: 

85 ```python 

86 class InMemoryEventBus: 

87 async def publish(self, event: DomainEvent) -> "Result[None, EventError]": 

88 for handler in self._handlers.get(type(event), []): 

89 await handler.handle(event) 

90 return Ok(None) 

91 

92 def subscribe(self, event_type, handler): 

93 self._handlers.setdefault(event_type, []).append(handler) 

94 ``` 

95 """ 

96 

97 async def publish(self, event: Any) -> Result[None, EventError]: 

98 r"""Publish an event to all subscribers. 

99 

100 Args: 

101 event: DomainEvent to publish. 

102 

103 Returns: 

104 Ok(None) when the event is successfully enqueued for dispatch. 

105 Err(EventError) when the event cannot be accepted 

106 (e.g.\ no handlers registered and the bus requires at least one). 

107 """ 

108 ... 

109 

110 def subscribe(self, event_type: type, handler: EventHandlerProtocol) -> None: 

111 """Subscribe a handler to an event type. 

112 

113 Args: 

114 event_type: Type of event to subscribe to. 

115 handler: Handler to call when event is published. 

116 """ 

117 ... 

118 

119 def unsubscribe(self, event_type: type, handler: EventHandlerProtocol) -> None: 

120 """Remove a handler subscription for an event type. 

121 

122 Args: 

123 event_type: Type of event to unsubscribe from. 

124 handler: Handler to remove. 

125 """ 

126 ... 

127 

128 

129@runtime_checkable 

130class EventMiddlewareProtocol(Protocol): 

131 """Protocol for event bus middleware. 

132 

133 Middleware intercepts event publication, allowing cross-cutting 

134 concerns like logging, metrics, or error handling to be applied 

135 transparently. 

136 

137 The middleware receives the event and a ``next_handler`` coroutine 

138 that invokes the next middleware (or the actual handlers). The 

139 middleware must call ``next_handler`` to continue the chain. 

140 

141 Example:: 

142 

143 class LoggingMiddleware: 

144 async def __call__(self, event: Any, next_handler: Any) -> None: 

145 logger.info("publishing", event_type=type(event).__name__) 

146 await next_handler(event) 

147 logger.info("published", event_type=type(event).__name__) 

148 """ 

149 

150 async def __call__(self, event: Any, next_handler: Any) -> None: 

151 """Process an event through this middleware. 

152 

153 Args: 

154 event: The domain event being published. 

155 next_handler: Coroutine to call to continue the middleware 

156 chain. Must be awaited for the event to reach handlers. 

157 """ 

158 ... 

159 

160 

161# --- Commands (CQRS) --- 

162 

163 

164@runtime_checkable 

165class CommandHandlerProtocol(Protocol): 

166 """Protocol for command handlers.""" 

167 

168 async def handle(self, command: Any) -> Any: 

169 """Handle a command. 

170 

171 Args: 

172 command: Command to handle. 

173 

174 Returns: 

175 Command result. 

176 """ 

177 ... 

178 

179 

180@runtime_checkable 

181class CommandBusProtocol(Protocol): 

182 """Protocol for command bus implementations. 

183 

184 Example: 

185 ```python 

186 class CommandBusProtocol: 

187 async def dispatch(self, command: Command) -> Any: 

188 handler = self._handlers[type(command)] 

189 return await handler.handle(command) 

190 ``` 

191 """ 

192 

193 async def dispatch(self, command: Any) -> Any: 

194 """Dispatch a command to its handler. 

195 

196 Args: 

197 command: Command to dispatch. 

198 

199 Returns: 

200 Result from the command handler. 

201 """ 

202 ... 

203 

204 

205# --- Queries (CQRS) --- 

206 

207 

208@runtime_checkable 

209class QueryHandlerProtocol(Protocol): 

210 """Protocol for query handlers.""" 

211 

212 async def handle(self, query: Any) -> Any: 

213 """Handle a query. 

214 

215 Args: 

216 query: Query to handle. 

217 

218 Returns: 

219 Query result. 

220 """ 

221 ... 

222 

223 

224@runtime_checkable 

225class QueryBusProtocol(Protocol): 

226 """Protocol for query bus implementations. 

227 

228 Example: 

229 ```python 

230 class QueryBusProtocol: 

231 async def execute(self, query: Query) -> Any: 

232 handler = self._handlers[type(query)] 

233 return await handler.handle(query) 

234 ``` 

235 """ 

236 

237 async def execute(self, query: Any) -> Any: 

238 """Execute a query through its handler. 

239 

240 Args: 

241 query: Query to execute. 

242 

243 Returns: 

244 Query result. 

245 """ 

246 ... 

247 

248 

249# --- Event Store --- 

250 

251 

252@runtime_checkable 

253class EventStoreProtocol(Protocol): 

254 """Protocol for event store implementations. 

255 

256 The event store persists and retrieves domain events. 

257 

258 Example: 

259 ```python 

260 class PostgresEventStore: 

261 async def append(self, stream_id, events, expected_version=None): 

262 async with self.db.transaction(): 

263 for event in events: 

264 await self.db.insert("events", event.to_dict()) 

265 ``` 

266 """ 

267 

268 async def append( 

269 self, 

270 stream_id: str, 

271 events: list[Any], 

272 expected_version: int | None = None, 

273 ) -> int: 

274 """Append events to a stream. 

275 

276 Args: 

277 stream_id: Unique stream identifier. 

278 events: List of events to append. 

279 expected_version: Expected stream version for optimistic concurrency. 

280 

281 Returns: 

282 New stream version. 

283 

284 Raises: 

285 ConcurrencyError: If expected version doesn't match. 

286 """ 

287 ... 

288 

289 async def read( 

290 self, 

291 stream_id: str, 

292 start: int = 0, 

293 count: int | None = None, 

294 ) -> list[Any]: 

295 """Read events from a stream. 

296 

297 Args: 

298 stream_id: Unique stream identifier. 

299 start: Starting position. 

300 count: Maximum events to read. 

301 

302 Returns: 

303 List of events. 

304 """ 

305 ... 

306 

307 async def read_all( 

308 self, 

309 position: int = 0, 

310 count: int | None = None, 

311 ) -> list[Any]: 

312 """Read events from all streams. 

313 

314 Args: 

315 position: Starting global position. 

316 count: Maximum events to read. 

317 

318 Returns: 

319 List of events. 

320 """ 

321 ... 

322 

323 

324@runtime_checkable 

325class SnapshotStoreProtocol(Protocol): 

326 """Protocol for aggregate snapshot storage.""" 

327 

328 async def save(self, aggregate_id: str, snapshot: Any, version: int) -> None: 

329 """Save an aggregate snapshot. 

330 

331 Args: 

332 aggregate_id: Aggregate identifier. 

333 snapshot: Aggregate state snapshot. 

334 version: Aggregate version at snapshot. 

335 """ 

336 ... 

337 

338 async def load(self, aggregate_id: str) -> tuple[Any, int] | None: 

339 """Load the latest snapshot. 

340 

341 Args: 

342 aggregate_id: Aggregate identifier. 

343 

344 Returns: 

345 Tuple of (snapshot, version) or None if not found. 

346 """ 

347 ... 

348 

349 

350# --- Event Sourced RepositoryProtocol --- 

351 

352 

353@runtime_checkable 

354class EventSourcedReadRepositoryProtocol(Protocol): 

355 """Protocol for read-only repository access.""" 

356 

357 async def get(self, aggregate_id: UUID | str) -> Any | None: 

358 """Load an aggregate by ID.""" 

359 ... 

360 

361 async def exists(self, aggregate_id: UUID | str) -> bool: 

362 """Check if an aggregate exists.""" 

363 ... 

364 

365 async def get_all( 

366 self, 

367 limit: int | None = None, 

368 offset: int | None = None, 

369 ) -> list[Any]: 

370 """Get all aggregates (paginated).""" 

371 ... 

372 

373 

374@runtime_checkable 

375class EventSourcedRepositoryProtocol(EventSourcedReadRepositoryProtocol, Protocol): 

376 """Protocol for event-sourced repositories (Read/Write).""" 

377 

378 async def save(self, aggregate: Any) -> None: 

379 """Save an aggregate.""" 

380 ... 

381 

382 

383@runtime_checkable 

384class AggregateFactoryProtocol(Protocol[TAggregate_co]): 

385 """Factory for creating aggregates.""" 

386 

387 def create(self, aggregate_id: UUID | str) -> TAggregate_co: 

388 """Create a new aggregate instance. 

389 

390 Args: 

391 aggregate_id: Aggregate ID. 

392 

393 Returns: 

394 New aggregate instance. 

395 """ 

396 ... 

397 

398 

399# --- Projections --- 

400 

401 

402@runtime_checkable 

403class ProjectionProtocol(Protocol): 

404 """ProjectionProtocol protocol for building read models from events.""" 

405 

406 def apply(self, event: Any) -> None: 

407 """Apply an event to the projection state. 

408 

409 Args: 

410 event: Domain event to apply. 

411 """ 

412 ... 

413 

414 

415@runtime_checkable 

416class PubSubProtocol(Protocol): 

417 """Protocol for publish/subscribe backends. 

418 

419 Implementations must support publishing messages to topics and 

420 subscribing handlers to receive messages from topics. 

421 """ 

422 

423 async def publish(self, topic: str, data: Any) -> None: 

424 """Publish *data* to *topic*.""" 

425 ... 

426 

427 async def subscribe( 

428 self, 

429 topic: str, 

430 handler: Any, 

431 ) -> None: 

432 """Subscribe *handler* to receive messages from *topic*.""" 

433 ... 

434 

435 async def unsubscribe(self, topic: str, handler: Any) -> None: 

436 """Remove *handler* subscription from *topic*.""" 

437 ... 

438 

439 

440@runtime_checkable 

441class IntegrationEventProtocol(Protocol): 

442 """Protocol formalising the bridge between domain events and transactional messaging. 

443 

444 Any event intended for cross-service communication should satisfy this 

445 protocol. Implementations carry all metadata required for reliable, 

446 idempotent delivery across bounded-context boundaries. 

447 

448 Attributes: 

449 event_id: Unique, stable identifier for this event instance (e.g. a UUID4 

450 string). Used by consumers for deduplication. 

451 event_type: The event class name or type discriminator string. Consumers 

452 use this to route / deserialise the ``payload``. 

453 source_service: The bounded context or service that emitted this event 

454 (e.g. ``"order-service"``). 

455 correlation_id: Optional distributed-trace correlation identifier that 

456 links this event to a broader request chain. ``None`` when tracing 

457 is not active. 

458 causation_id: Optional identifier of the event or command that directly 

459 caused this event to be emitted. ``None`` for root events. 

460 payload: The event data as a JSON-serialisable dictionary. Must not 

461 contain non-serialisable types (e.g. ``datetime`` objects must be 

462 ISO-formatted strings). 

463 occurred_at: UTC timestamp recording when the event occurred in the 

464 source service. 

465 

466 Example:: 

467 

468 class UserRegisteredIntegrationEvent: 

469 event_id: str = field(default_factory=lambda: str(uuid4())) 

470 event_type: str = "UserRegistered" 

471 source_service: str = "identity-service" 

472 correlation_id: str | None = None 

473 causation_id: str | None = None 

474 payload: dict[str, Any] = field(default_factory=dict) 

475 occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

476 """ 

477 

478 event_id: str 

479 event_type: str 

480 source_service: str 

481 correlation_id: str | None 

482 causation_id: str | None 

483 payload: dict[str, Any] 

484 occurred_at: datetime 

485 

486 

487@runtime_checkable 

488class WebhookSignatureVerifierProtocol(Protocol): 

489 """Verifies the authenticity of inbound webhook payloads. 

490 

491 Implementations must provide both a verification method and a signature 

492 computation method so callers can independently validate or generate 

493 signatures without exposing the underlying algorithm. 

494 

495 The canonical implementation is HMAC-SHA256, but any MAC or asymmetric 

496 scheme that fulfils this protocol is acceptable. 

497 

498 Typical usage:: 

499 

500 verifier = HMACWebhookVerifier() 

501 if not verifier.verify(payload=body, signature=sig_header, secret=secret): 

502 raise PermissionError("Invalid webhook signature") 

503 """ 

504 

505 def verify( 

506 self, 

507 payload: bytes, 

508 signature: str, 

509 secret: str, 

510 ) -> bool: 

511 """Return ``True`` when the signature matches the payload. 

512 

513 Implementations must use a constant-time comparison to prevent 

514 timing side-channel attacks. 

515 

516 Args: 

517 payload: Raw request body bytes. 

518 signature: Signature string as received in the request header 

519 (may include algorithm prefix such as ``"sha256=..."``). 

520 secret: Shared secret used to compute the expected signature. 

521 

522 Returns: 

523 ``True`` if the signature is valid, ``False`` otherwise. 

524 """ 

525 ... 

526 

527 def compute_signature(self, payload: bytes, secret: str) -> str: 

528 """Compute the expected signature for *payload* using *secret*. 

529 

530 Args: 

531 payload: Raw request body bytes. 

532 secret: Shared secret. 

533 

534 Returns: 

535 Hex-encoded signature string in the same format that 

536 :meth:`verify` expects as input. 

537 """ 

538 ... 

539 

540 

541__all__ = [ 

542 "AggregateFactoryProtocol", 

543 "CommandBusProtocol", 

544 "CommandHandlerProtocol", 

545 "DomainEventPublisherProtocol", 

546 "EventBusProtocol", 

547 "EventHandlerProtocol", 

548 "EventMiddlewareProtocol", 

549 "EventSourcedReadRepositoryProtocol", 

550 "EventSourcedRepositoryProtocol", 

551 "EventStoreProtocol", 

552 "IntegrationEventProtocol", 

553 "MultiEventHandlerProtocol", 

554 "ProjectionProtocol", 

555 "PubSubProtocol", 

556 "QueryBusProtocol", 

557 "QueryHandlerProtocol", 

558 "SnapshotStoreProtocol", 

559 "WebhookSignatureVerifierProtocol", 

560]