Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/email/email_senders.py: 39%

76 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:32 +0000

1import smtplib 1a

2import typing 1a

3from abc import ABC, abstractmethod 1a

4from dataclasses import dataclass 1a

5from email import message 1a

6from email.utils import formatdate 1a

7from uuid import uuid4 1a

8 

9from html2text import html2text 1a

10 

11from mealie.services._base_service import BaseService 1a

12 

13SMTP_TIMEOUT = 10 1a

14"""Timeout in seconds for SMTP connection""" 1a

15 

16 

17@dataclass(slots=True) 1a

18class EmailOptions: 1a

19 host: str 1a

20 port: int 1a

21 username: str | None = None 1a

22 password: str | None = None 1a

23 tls: bool = False 1a

24 ssl: bool = False 1a

25 

26 

27@dataclass(slots=True) 1a

28class SMTPResponse: 1a

29 success: bool 1a

30 message: str 1a

31 errors: typing.Any 1a

32 

33 

34@dataclass(slots=True) 1a

35class Message: 1a

36 subject: str 1a

37 html: str 1a

38 mail_from_name: str 1a

39 mail_from_address: str 1a

40 

41 def send(self, to: str, smtp: EmailOptions) -> SMTPResponse: 1a

42 msg = message.EmailMessage() 

43 msg["Subject"] = self.subject 

44 msg["From"] = f"{self.mail_from_name} <{self.mail_from_address}>" 

45 msg["To"] = to 

46 msg["Date"] = formatdate(localtime=True) 

47 msg.add_alternative(html2text(self.html), subtype="plain") 

48 msg.add_alternative(self.html, subtype="html") 

49 

50 try: 

51 message_id = f"<{uuid4()}@{self.mail_from_address.split('@')[1]}>" 

52 except IndexError: 

53 # this should never happen with a valid email address, 

54 # but we let the SMTP server handle it instead of raising it here 

55 message_id = f"<{uuid4()}@{self.mail_from_address}>" 

56 

57 msg["Message-ID"] = message_id 

58 msg["MIME-Version"] = "1.0" 

59 

60 if smtp.ssl: 

61 with smtplib.SMTP_SSL(smtp.host, smtp.port, timeout=SMTP_TIMEOUT) as server: 

62 if smtp.username and smtp.password: 

63 server.login(smtp.username, smtp.password) 

64 

65 errors = server.send_message(msg) 

66 else: 

67 with smtplib.SMTP(smtp.host, smtp.port, timeout=SMTP_TIMEOUT) as server: 

68 if smtp.tls: 

69 server.starttls() 

70 if smtp.username and smtp.password: 

71 server.login(smtp.username, smtp.password) 

72 errors = server.send_message(msg) 

73 

74 return SMTPResponse(errors == {}, "Message Sent", errors=errors) 

75 

76 

77class ABCEmailSender(ABC): 1a

78 @abstractmethod 1a

79 def send(self, email_to: str, subject: str, html: str) -> bool: ... 79 ↛ exitline 79 didn't return from function 'send' because 1a

80 

81 

82class DefaultEmailSender(ABCEmailSender, BaseService): 1a

83 """ 

84 DefaultEmailSender is the default email sender for Mealie. It uses the SMTP settings 

85 from the config file to send emails via the python standard library. It supports 

86 both TLS and SSL connections. 

87 """ 

88 

89 def send(self, email_to: str, subject: str, html: str) -> bool: 1a

90 if self.settings.SMTP_FROM_EMAIL is None or self.settings.SMTP_FROM_NAME is None: 

91 raise ValueError("SMTP_FROM_EMAIL and SMTP_FROM_NAME must be set in the config file.") 

92 

93 message = Message( 

94 subject=subject, 

95 html=html, 

96 mail_from_name=self.settings.SMTP_FROM_NAME, 

97 mail_from_address=self.settings.SMTP_FROM_EMAIL, 

98 ) 

99 

100 if self.settings.SMTP_HOST is None or self.settings.SMTP_PORT is None: 

101 raise ValueError("SMTP_HOST, SMTP_PORT must be set in the config file.") 

102 

103 smtp_options = EmailOptions( 

104 self.settings.SMTP_HOST, 

105 int(self.settings.SMTP_PORT), 

106 tls=self.settings.SMTP_AUTH_STRATEGY.upper() == "TLS" if self.settings.SMTP_AUTH_STRATEGY else False, 

107 ssl=self.settings.SMTP_AUTH_STRATEGY.upper() == "SSL" if self.settings.SMTP_AUTH_STRATEGY else False, 

108 ) 

109 

110 if self.settings.SMTP_USER: 

111 smtp_options.username = self.settings.SMTP_USER 

112 if self.settings.SMTP_PASSWORD: 

113 smtp_options.password = self.settings.SMTP_PASSWORD 

114 

115 response = message.send(to=email_to, smtp=smtp_options) 

116 self.logger.debug(f"send email result: {response}") 

117 

118 if not response.success: 

119 self.logger.error(f"send email error: {response}") 

120 

121 return response.success