Coverage for extensions/smtp.py: 97.89%

1import logging 

2import smtplib 

3import ssl 

4from collections.abc import Iterable 

5from dataclasses import dataclass 

6from email.message import EmailMessage 

7from email.mime.multipart import MIMEMultipart 

8from email.mime.text import MIMEText 

9 

10from fastapi import FastAPI 

11 

12from core.config import settings 

13from utils.custom_exception import SMTPNotConfiguredException 

14 

15logger = logging.getLogger("smtp") 

16 

17 

18@dataclass(frozen=True) 

19class SMTPSettings: 

20 enabled: bool 

21 host: str 

22 port: int 

23 username: str | None 

24 password: str | None 

25 from_email: str | None 

26 from_name: str 

27 encryption: str 

28 

29 

30class SMTPMailer: 

31 def __init__(self, cfg: SMTPSettings): 

32 self._cfg = cfg 

33 

34 @property 

35 def enabled(self) -> bool: 

36 return self._cfg.enabled 

37 

38 def _validate(self) -> None: 

39 if not self._cfg.enabled: 

40 raise SMTPNotConfiguredException("SMTP is disabled") 

41 missing = [] 

42 if not self._cfg.host: 

43 missing.append("SMTP_HOST") 

44 if not self._cfg.port: 

45 missing.append("SMTP_PORT") 

46 if not self._cfg.username: 

47 missing.append("SMTP_USERNAME/SMTP_USER") 

48 if not self._cfg.password: 

49 missing.append("SMTP_PASSWORD") 

50 if not self._cfg.from_email: 

51 missing.append("SMTP_FROM_EMAIL/SMTP_FROM") 

52 if missing: 

53 raise SMTPNotConfiguredException(f"Missing SMTP settings: {', '.join(missing)}") 

54 

55 enc = (self._cfg.encryption or "").strip().lower() 

56 if enc not in {"tls", "ssl", "none"}: 

57 raise SMTPNotConfiguredException("SMTP_ENCRYPTION must be tls, ssl, or none") 

58 

59 def _open(self, timeout: int = 30) -> smtplib.SMTP: 

60 """ 

61 Create a new SMTP connection per send. 

62 This avoids keeping long-lived connections in the web process. 

63 """ 

64 self._validate() 

65 

66 enc = self._cfg.encryption.strip().lower() 

67 context = ssl.create_default_context() 

68 

69 if enc == "ssl": 

70 client: smtplib.SMTP = smtplib.SMTP_SSL( 

71 self._cfg.host, self._cfg.port, timeout=timeout, context=context 

72 ) 

73 else: 

74 client = smtplib.SMTP(self._cfg.host, self._cfg.port, timeout=timeout) 

75 

76 try: 

77 client.ehlo() 

78 if enc == "tls": 

79 client.starttls(context=context) 

80 client.ehlo() 

81 if self._cfg.username and self._cfg.password: 

82 client.login(self._cfg.username, self._cfg.password) 

83 return client 

84 except Exception as e: 

85 logger.error( 

86 "SMTP connection failed: host=%s port=%s encryption=%s error=%s", 

87 self._cfg.host, 

88 self._cfg.port, 

89 enc, 

90 e, 

91 exc_info=True, 

92 ) 

93 try: 

94 client.quit() 

95 except Exception: 

96 pass 

97 raise 

98 

99 def send_text( 

100 self, 

101 *, 

102 to_emails: Iterable[str], 

103 subject: str, 

104 body: str, 

105 html_body: str | None = None, 

106 from_email: str | None = None, 

107 from_name: str | None = None, 

108 timeout: int = 30, 

109 ) -> None: 

110 self._validate() 

111 

112 sender_email = from_email or self._cfg.from_email 

113 sender_name = from_name or self._cfg.from_name 

114 

115 # If HTML body is provided, create multipart message 

116 if html_body: 

117 msg = MIMEMultipart("alternative") 

118 msg["Subject"] = subject 

119 msg["From"] = f"{sender_name} <{sender_email}>" 

120 msg["To"] = ", ".join(list(to_emails)) 

121 

122 # Add plain text and HTML parts 

123 part1 = MIMEText(body, "plain", "utf-8") 

124 part2 = MIMEText(html_body, "html", "utf-8") 

125 msg.attach(part1) 

126 msg.attach(part2) 

127 else: 

128 # Plain text only 

129 msg = EmailMessage() 

130 msg["Subject"] = subject 

131 msg["From"] = f"{sender_name} <{sender_email}>" 

132 msg["To"] = ", ".join(list(to_emails)) 

133 msg.set_content(body) 

134 

135 with self._open(timeout=timeout) as client: 

136 client.send_message(msg) 

137 

138 

139def build_smtp_settings() -> SMTPSettings: 

140 return SMTPSettings( 

141 enabled=bool(settings.SMTP_ENABLE), 

142 host=(settings.SMTP_HOST or "").strip(), 

143 port=int(settings.SMTP_PORT), 

144 username=(settings.SMTP_USERNAME or None), 

145 password=(settings.SMTP_PASSWORD or None), 

146 from_email=(settings.SMTP_FROM_EMAIL or None), 

147 from_name=(settings.SMTP_FROM_NAME or "Docker Fullstack Template"), 

148 encryption=(settings.SMTP_ENCRYPTION or "tls"), 

149 ) 

150 

151 

152# Global singleton instance 

153_SMTP_MAILER: SMTPMailer | None = None 

154 

155 

156def get_mailer() -> SMTPMailer: 

157 """ 

158 Get SMTP mailer singleton instance. 

159 Lazy initialization on first access. 

160 """ 

161 global _SMTP_MAILER 

162 if _SMTP_MAILER is None: 

163 cfg = build_smtp_settings() 

164 _SMTP_MAILER = SMTPMailer(cfg) 

165 

166 if cfg.enabled: 

167 logger.info( 

168 "SMTP enabled: host=%s port=%s encryption=%s from=%s", 

169 cfg.host, 

170 cfg.port, 

171 (cfg.encryption or "").lower(), 

172 cfg.from_email, 

173 ) 

174 else: 

175 logger.info("SMTP disabled") 

176 

177 return _SMTP_MAILER 

178 

179 

180def add_smtp(app: FastAPI) -> None: 

181 """ 

182 Initialize SMTP mailer and register to app.state. 

183 """ 

184 mailer = get_mailer() 

185 app.state.smtp = mailer