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

1# lexigram-contracts/src/lexigram/contracts/notification/protocols.py 

2"""Notification channel protocols.""" 

3 

4from __future__ import annotations 

5 

6from typing import TYPE_CHECKING, Protocol, runtime_checkable 

7 

8if TYPE_CHECKING: 

9 from lexigram.contracts.core.health import HealthCheckResult 

10 from lexigram.contracts.core.result import Result 

11 from lexigram.contracts.mailer.types import MessageDeliveryReceipt 

12 from lexigram.contracts.notification.errors import NotificationError 

13 from lexigram.contracts.notification.types import PushMessage, SMSMessage 

14 

15 

16@runtime_checkable 

17class SMSChannelProtocol(Protocol): 

18 """Structural protocol for SMS backends (Twilio, Vonage, etc.). 

19 

20 ``send()`` returns ``Ok(receipt)`` on acceptance or ``Err(NotificationError)`` 

21 for expected failures. Infrastructure failures must be raised as exceptions. 

22 """ 

23 

24 async def send( 

25 self, message: SMSMessage 

26 ) -> Result[MessageDeliveryReceipt, NotificationError]: 

27 """Send an SMS message. 

28 

29 Args: 

30 message: The SMS to deliver. 

31 

32 Returns: 

33 ``Ok(MessageDeliveryReceipt)`` or ``Err(NotificationError)``. 

34 """ 

35 ... 

36 

37 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

38 """Check SMS backend connectivity.""" 

39 ... 

40 

41 

42@runtime_checkable 

43class PushChannelProtocol(Protocol): 

44 """Structural protocol for push notification backends (FCM, APNS, etc.). 

45 

46 ``send()`` delivers to one or more device tokens. ``send_batch()`` delivers 

47 to multiple tokens in one API call where the provider supports it. 

48 """ 

49 

50 async def send( 

51 self, message: PushMessage 

52 ) -> Result[MessageDeliveryReceipt, NotificationError]: 

53 """Send a push notification to one or more device tokens. 

54 

55 Args: 

56 message: The push notification to deliver. 

57 

58 Returns: 

59 ``Ok(MessageDeliveryReceipt)`` or ``Err(NotificationError)``. 

60 """ 

61 ... 

62 

63 async def send_batch( 

64 self, messages: list[PushMessage] 

65 ) -> list[Result[MessageDeliveryReceipt, NotificationError]]: 

66 """Send push notifications in bulk. 

67 

68 Args: 

69 messages: List of push notifications to deliver. 

70 

71 Returns: 

72 List of ``Result`` values, one per message, preserving order. 

73 """ 

74 ... 

75 

76 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

77 """Check push backend connectivity.""" 

78 ... 

79 

80 

81__all__ = ["PushChannelProtocol", "SMSChannelProtocol"]