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

1"""Outbox pattern protocols for reliable event publishing. 

2 

3Defines the contracts that both in-memory (core) and database-backed 

4(lexigram-sql / lexigram-messaging) outbox implementations must satisfy. 

5 

6The Outbox pattern stores domain events in the same transaction as 

7business data, then relays them asynchronously to the event bus, 

8guaranteeing at-least-once delivery. 

9""" 

10 

11from __future__ import annotations 

12 

13from enum import StrEnum 

14from typing import Any, Protocol, runtime_checkable 

15 

16__all__ = [ 

17 "OutboxBackendProtocol", 

18 "OutboxEntryProtocol", 

19 "OutboxRelayProtocol", 

20 "OutboxStatus", 

21] 

22 

23 

24class OutboxStatus(StrEnum): 

25 """Lifecycle status of an outbox entry.""" 

26 

27 PENDING = "pending" 

28 PUBLISHED = "published" 

29 FAILED = "failed" 

30 

31 

32@runtime_checkable 

33class OutboxEntryProtocol(Protocol): 

34 """Protocol describing a single persisted outbox entry. 

35 

36 Concrete implementations may be dataclasses, ORM models, or plain 

37 dicts — anything that exposes these attributes satisfies the protocol. 

38 """ 

39 

40 @property 

41 def entry_id(self) -> str: 

42 """Unique identifier for this outbox entry.""" 

43 ... 

44 

45 @property 

46 def event(self) -> Any: 

47 """The domain event payload to be published.""" 

48 ... 

49 

50 @property 

51 def status(self) -> OutboxStatus: 

52 """Current lifecycle status of the entry.""" 

53 ... 

54 

55 @property 

56 def error(self) -> str | None: 

57 """Error message from the last failed publish attempt, if any.""" 

58 ... 

59 

60 

61@runtime_checkable 

62class OutboxBackendProtocol(Protocol): 

63 """Protocol for the outbox store — write side of the outbox pattern. 

64 

65 Implementations are responsible for persisting entries and updating 

66 their status. In production this maps to a database table; in tests 

67 an in-memory deque suffices. 

68 

69 Note: For SQL-backed outbox storage, use 

70 ``lexigram.contracts.data.outbox.OutboxStoreProtocol``. 

71 """ 

72 

73 async def store_event(self, event: Any) -> OutboxEntryProtocol: 

74 """Persist a domain event as a new PENDING outbox entry. 

75 

76 Args: 

77 event: Domain event to store. 

78 

79 Returns: 

80 The newly created outbox entry. 

81 """ 

82 ... 

83 

84 async def get_pending(self, *, limit: int = 100) -> list[OutboxEntryProtocol]: 

85 """Retrieve up to *limit* entries with PENDING status. 

86 

87 Args: 

88 limit: Maximum number of entries to return. 

89 

90 Returns: 

91 List of pending outbox entries, oldest first. 

92 """ 

93 ... 

94 

95 async def mark_published(self, entry_id: str) -> None: 

96 """Mark an entry as successfully PUBLISHED. 

97 

98 Args: 

99 entry_id: Identifier of the entry to update. 

100 """ 

101 ... 

102 

103 async def mark_failed(self, entry_id: str, error: str) -> None: 

104 """Mark an entry as FAILED and record the error message. 

105 

106 Args: 

107 entry_id: Identifier of the entry to update. 

108 error: Human-readable description of the failure. 

109 """ 

110 ... 

111 

112 

113@runtime_checkable 

114class OutboxRelayProtocol(Protocol): 

115 """Protocol for the outbox relay — read/dispatch side of the pattern. 

116 

117 The relay polls the store for pending entries and publishes them to 

118 the event bus, updating entry status after each attempt. 

119 """ 

120 

121 async def process_pending(self) -> tuple[int, int]: 

122 """Process all currently pending outbox entries. 

123 

124 Returns: 

125 Tuple of ``(published_count, failed_count)``. 

126 """ 

127 ...