Coverage for extensions/smtp.py: 98.04%

1import ssl 

2import smtplib 

3import logging 

4from fastapi import FastAPI 

5from core.config import settings 

6from dataclasses import dataclass 

7from email.mime.text import MIMEText 

8from typing import Iterable, Optional 

9from email.message import EmailMessage 

10from email.mime.multipart import MIMEMultipart 

11from utils.custom_exception import SMTPNotConfiguredException 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16@dataclass(frozen=True) 

17class SMTPSettings: 

18 enabled: bool 

19 host: str 

20 port: int 

21 username: Optional[str] 

22 password: Optional[str] 

23 from_email: Optional[str] 

24 from_name: str 

25 encryption: str 

26 

27 

28class SMTPMailer: 

29 def __init__(self, cfg: SMTPSettings): 

30 self._cfg = cfg 

31 

32 @property 

33 def enabled(self) -> bool: 

34 return self._cfg.enabled 

35 

36 def _validate(self) -> None: 

37 if not self._cfg.enabled: 

38 raise SMTPNotConfiguredException("SMTP is disabled") 

39 missing = [] 

40 if not self._cfg.host: 

41 missing.append("SMTP_HOST") 

42 if not self._cfg.port: 

43 missing.append("SMTP_PORT") 

44 if not self._cfg.username: 

45 missing.append("SMTP_USERNAME/SMTP_USER") 

46 if not self._cfg.password: 

47 missing.append("SMTP_PASSWORD") 

48 if not self._cfg.from_email: 

49 missing.append("SMTP_FROM_EMAIL/SMTP_FROM") 

50 if missing: 

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

52 

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

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

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

56 

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

58 """ 

59 Create a new SMTP connection per send. 

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

61 """ 

62 self._validate() 

63 

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

65 context = ssl.create_default_context() 

66 

67 if enc == "ssl": 

68 client: smtplib.SMTP = smtplib.SMTP_SSL(self._cfg.host, self._cfg.port, timeout=timeout, context=context) 

69 else: 

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

71 

72 try: 

73 client.ehlo() 

74 if enc == "tls": 

75 client.starttls(context=context) 

76 client.ehlo() 

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

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

79 return client 

80 except Exception: 

81 try: 

82 client.quit() 

83 except Exception: 

84 pass 

85 raise 

86 

87 def send_text( 

88 self, 

89 *, 

90 to_emails: Iterable[str], 

91 subject: str, 

92 body: str, 

93 html_body: Optional[str] = None, 

94 from_email: Optional[str] = None, 

95 from_name: Optional[str] = None, 

96 timeout: int = 30, 

97 ) -> None: 

98 self._validate() 

99 

100 sender_email = from_email or self._cfg.from_email 

101 sender_name = from_name or self._cfg.from_name 

102 

103 # If HTML body is provided, create multipart message 

104 if html_body: 

105 msg = MIMEMultipart("alternative") 

106 msg["Subject"] = subject 

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

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

109 

110 # Add plain text and HTML parts 

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

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

113 msg.attach(part1) 

114 msg.attach(part2) 

115 else: 

116 # Plain text only 

117 msg = EmailMessage() 

118 msg["Subject"] = subject 

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

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

121 msg.set_content(body) 

122 

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

124 client.send_message(msg) 

125 

126 

127def build_smtp_settings() -> SMTPSettings: 

128 return SMTPSettings( 

129 enabled=bool(settings.SMTP_ENABLE), 

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

131 port=int(settings.SMTP_PORT), 

132 username=(settings.SMTP_USERNAME or None), 

133 password=(settings.SMTP_PASSWORD or None), 

134 from_email=(settings.SMTP_FROM_EMAIL or None), 

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

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

137 ) 

138 

139 

140# Global singleton instance 

141_SMTP_MAILER: Optional[SMTPMailer] = None 

142 

143 

144def get_mailer() -> SMTPMailer: 

145 """ 

146 Get SMTP mailer singleton instance. 

147 Lazy initialization on first access. 

148 """ 

149 global _SMTP_MAILER 

150 if _SMTP_MAILER is None: 

151 cfg = build_smtp_settings() 

152 _SMTP_MAILER = SMTPMailer(cfg) 

153 

154 if cfg.enabled: 

155 logger.info( 

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

157 cfg.host, 

158 cfg.port, 

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

160 cfg.from_email, 

161 ) 

162 else: 

163 logger.info("SMTP disabled") 

164 

165 return _SMTP_MAILER 

166 

167 

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

169 """ 

170 Initialize SMTP mailer and register to app.state. 

171 """ 

172 mailer = get_mailer() 

173 app.state.smtp = mailer