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

7 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-08-19 05:41 +0800

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

2"""MailerProtocol — structural contract for email backends.""" 

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.errors import MailerError 

12 from lexigram.contracts.mailer.types import EmailMessage, MessageDeliveryReceipt 

13 

14 

15@runtime_checkable 

16class MailerProtocol(Protocol): 

17 """Structural protocol for email delivery backends. 

18 

19 All mailer backends (SMTP, SendGrid, SES, etc.) must satisfy this 

20 protocol for Named DI resolution. ``send()`` accepts a fully-formed 

21 :class:`~lexigram.contracts.mailer.types.EmailMessage` and 

22 returns ``Ok(receipt)`` on success or ``Err(MailerError)`` for expected 

23 delivery failures. Infrastructure failures (authentication, network 

24 timeout) must be raised as exceptions, not wrapped in ``Err``. 

25 """ 

26 

27 async def send( 

28 self, message: EmailMessage 

29 ) -> Result[MessageDeliveryReceipt, MailerError]: 

30 """Send an email message. 

31 

32 Args: 

33 message: The email to deliver. 

34 

35 Returns: 

36 ``Ok(MessageDeliveryReceipt)`` on acceptance by the backend. 

37 ``Err(MailerError)`` for expected delivery failures. 

38 """ 

39 ... 

40 

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

42 """Check backend connectivity. 

43 

44 Args: 

45 timeout: Maximum seconds to wait for a response. 

46 

47 Returns: 

48 :class:`~lexigram.contracts.core.HealthCheckResult` with status details. 

49 """ 

50 ... 

51 

52 

53__all__ = ["MailerProtocol"]