Coverage for webapp/utils/emailer.py: 95%

1from typing import List, Union 

2import smtplib 

3from email.mime.text import MIMEText 

4from email.mime.multipart import MIMEMultipart 

5import os 

6from collections import namedtuple 

7import logging 

8from flask import render_template 

9from threading import Thread 

10 

11SMTPConfig = namedtuple( 

12 "SMTPConfig", ["host", "port", "username", "password", "domain"] 

13) 

14 

15logger = logging.getLogger("emailer") 

16 

17 

18class EmailerError(Exception): 

19 """Custom exception for emailer errors""" 

20 

21 pass 

22 

23 

24class Emailer: 

25 def __init__(self, smtp_config: SMTPConfig): 

26 self.smtp_config = smtp_config 

27 self.is_configured = self._validate_config() 

28 

29 if self.is_configured: 

30 logger.info("Emailer initialized with SMTP configuration.") 

31 else: 

32 logger.warning( 

33 "Emailer disabled: " 

34 "SMTP configuration is missing or incomplete. " 

35 "Set SMTP_HOST, SMTP_USER, and SMTP_PASSWORD environment" 

36 " variables to enable email sending." 

37 ) 

38 

39 def _validate_config(self) -> bool: 

40 if not self.smtp_config: 

41 return False 

42 

43 return all( 

44 [ 

45 self.smtp_config.host, 

46 self.smtp_config.username, 

47 self.smtp_config.password, 

48 ] 

49 ) 

50 

51 def _create_message( 

52 self, 

53 subject: str, 

54 body: str, 

55 to_email: str | List[str], 

56 body_type: str = "plain", 

57 ) -> MIMEMultipart: 

58 """Create email message with proper headers""" 

59 msg = MIMEMultipart() 

60 # if username has the email format, use it as the sender 

61 if "@" in self.smtp_config.username: 

62 msg["From"] = self.smtp_config.username 

63 else: 

64 msg["From"] = ( 

65 f"noreply+{self.smtp_config.username}" 

66 f"@{self.smtp_config.domain}" 

67 ) 

68 msg["Subject"] = subject 

69 

70 if isinstance(to_email, list): 

71 msg["To"] = ", ".join(to_email) 

72 else: 

73 msg["To"] = to_email 

74 msg.attach(MIMEText(body, body_type)) 

75 return msg 

76 

77 def _send( 

78 self, 

79 subject: str, 

80 body: str, 

81 to_email: Union[str, List[str]], 

82 body_type: str = "plain", 

83 ): 

84 if not self.is_configured: 

85 return 

86 try: 

87 msg = self._create_message(subject, body, to_email, body_type) 

88 

89 with smtplib.SMTP( 

90 self.smtp_config.host, self.smtp_config.port 

91 ) as server: 

92 server.starttls() 

93 server.login( 

94 self.smtp_config.username, self.smtp_config.password 

95 ) 

96 

97 to_addrs = ( 

98 to_email if isinstance(to_email, list) else [to_email] 

99 ) 

100 server.send_message(msg, to_addrs=to_addrs) 

101 

102 except smtplib.SMTPException as e: 

103 logger.error(f"SMTP error occurred: {e}") 

104 raise EmailerError(f"Failed to send email: SMTP error - {e}") 

105 except Exception as e: 

106 logger.error(f"Unexpected error sending email: {e}") 

107 raise EmailerError(f"Failed to send email: {e}") 

108 

109 def send_email( 

110 self, subject: str, body: str, to_email: Union[str, List[str]] 

111 ): 

112 Thread(target=self._send, args=(subject, body, to_email)).start() 

113 

114 def send_email_template( 

115 self, to_email: str, subject: str, template_path: str, context: dict 

116 ): 

117 body = render_template(template_path, **context) 

118 Thread( 

119 target=self._send, args=(subject, body, to_email, "html") 

120 ).start() 

121 

122 

123smtp_config = SMTPConfig( 

124 host=os.getenv("SMTP_HOST", None), 

125 port=int(os.getenv("SMTP_PORT", 587)), 

126 username=os.getenv("SMTP_USER", None), 

127 password=os.getenv("SMTP_PASSWORD", None), 

128 domain=os.getenv("SMTP_DOMAIN", "canonical.com"), 

129) 

130 

131 

132def get_emailer() -> Emailer: 

133 smtp_config = SMTPConfig( 

134 host=os.getenv("SMTP_HOST", None), 

135 port=int(os.getenv("SMTP_PORT", 587)), 

136 username=os.getenv("SMTP_USER", None), 

137 password=os.getenv("SMTP_PASSWORD", None), 

138 domain=os.getenv("SMTP_DOMAIN", "canonical.com"), 

139 ) 

140 return Emailer(smtp_config=smtp_config)