Coverage for src / lexigram / contracts / notification / delivery.py: 100%

1"""DeliveryStoreProtocol — contract for persistent notification delivery tracking.""" 

2 

3from __future__ import annotations 

4 

5from enum import StrEnum 

6from typing import Any, Protocol, runtime_checkable 

7 

8 

9class DeliveryStatus(StrEnum): 

10 """States in the delivery lifecycle.""" 

11 

12 PENDING = "pending" 

13 DELIVERED = "delivered" 

14 RETRYING = "retrying" 

15 FAILED = "failed" 

16 

17 

18@runtime_checkable 

19class DeliveryStoreProtocol(Protocol): 

20 """Protocol for persisting notification delivery attempts. 

21 

22 Implementations record each send attempt and its outcome so that 

23 delivery history is preserved for auditing, alerting, and dead-letter 

24 inspection. A single delivery operation may generate multiple attempt 

25 records when retry logic is in play. 

26 """ 

27 

28 async def record_attempt( 

29 self, 

30 delivery_id: str, 

31 recipient: str, 

32 subject: str, 

33 attempt_number: int, 

34 ) -> None: 

35 """Record a delivery attempt. 

36 

37 Args: 

38 delivery_id: Stable identifier for the overall delivery operation. 

39 recipient: Comma-separated list of recipient addresses. 

40 subject: Message subject line. 

41 attempt_number: 1-based attempt counter within this delivery. 

42 """ 

43 ... 

44 

45 async def create_pending(self, message: Any) -> str: 

46 """Record a new pending delivery and return its delivery_id. 

47 

48 Args: 

49 message: The message being sent. 

50 

51 Returns: 

52 A unique delivery_id string for tracking this delivery. 

53 """ 

54 ... 

55 

56 async def mark_delivered(self, delivery_id: str) -> None: 

57 """Mark a delivery as successfully delivered. 

58 

59 Args: 

60 delivery_id: Identifier for the delivery operation. 

61 """ 

62 ... 

63 

64 async def get_retry_count(self, delivery_id: str) -> int: 

65 """Return the number of delivery attempts made so far. 

66 

67 Args: 

68 delivery_id: Identifier for the delivery operation. 

69 

70 Returns: 

71 Current attempt count. 

72 """ 

73 ... 

74 

75 async def increment_retry(self, delivery_id: str) -> int: 

76 """Increment the attempt counter and return the new value. 

77 

78 Args: 

79 delivery_id: Identifier for the delivery operation. 

80 

81 Returns: 

82 New attempt count after incrementing. 

83 """ 

84 ... 

85 

86 async def schedule_retry(self, delivery_id: str, delay_seconds: float) -> None: 

87 """Schedule a retry after ``delay_seconds``. 

88 

89 Args: 

90 delivery_id: Identifier for the delivery operation. 

91 delay_seconds: Seconds to wait before retrying. 

92 """ 

93 ... 

94 

95 async def mark_failed( 

96 self, 

97 delivery_id: str, 

98 reason: str = "", 

99 final: bool = True, 

100 ) -> None: 

101 """Mark a delivery attempt as failed. 

102 

103 Args: 

104 delivery_id: Identifier for the delivery operation. 

105 reason: Human-readable failure description. 

106 final: ``True`` when all retry attempts are exhausted and no further 

107 sends will be attempted. ``False`` for intermediate failures 

108 where retry is still pending. 

109 """ 

110 ... 

111 

112 

113__all__ = ["DeliveryStatus", "DeliveryStoreProtocol"]