Coverage for src / lexigram / admin / services / notifications / sender.py: 50%

1"""Email sending utilities for admin notifications.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.contracts.mailer import EmailMessage 

8from lexigram.logging import get_logger 

9 

10if TYPE_CHECKING: 

11 from lexigram.admin.services.notifications.models import NotificationRecipient 

12 from lexigram.contracts.mailer.protocols import MailerProtocol 

13 

14logger = get_logger(__name__) 

15 

16 

17class EmailSender: 

18 """Handles sending of notification emails via a MailerProtocol backend.""" 

19 

20 def __init__( 

21 self, 

22 mailer: MailerProtocol | None = None, 

23 from_email: str = "admin@example.com", 

24 from_name: str = "Admin System", 

25 ): 

26 """Initialize email sender. 

27 

28 Args: 

29 mailer: MailerProtocol backend for email delivery; no-ops when None. 

30 from_email: Default from email address. 

31 from_name: Default from name. 

32 """ 

33 self.mailer = mailer 

34 self.from_email = from_email 

35 self.from_name = from_name 

36 

37 async def send_email( 

38 self, 

39 recipient: NotificationRecipient, 

40 subject: str, 

41 body: str, 

42 html_body: str | None = None, 

43 ) -> None: 

44 """Send email to recipient. 

45 

46 Args: 

47 recipient: Email recipient. 

48 subject: Email subject. 

49 body: Plain text body. 

50 html_body: Optional HTML body. 

51 

52 Raises: 

53 RuntimeError: If email sending fails. 

54 """ 

55 if self.mailer: 

56 message = EmailMessage( 

57 to=[recipient.email], 

58 subject=subject, 

59 body=body, 

60 html_body=html_body, 

61 from_email=self.from_email, 

62 from_name=self.from_name, 

63 ) 

64 result = await self.mailer.send(message) 

65 if result.is_err(): 

66 raise RuntimeError(str(result.unwrap_err())) 

67 else: 

68 logger.info( 

69 "notification_email_skipped", 

70 to=recipient.email, 

71 subject=subject, 

72 ) 

73 

74 

75__all__ = ["EmailSender"]