Coverage for gws-app/gws/plugin/email_helper/__init__.py: 0%

1"""Email sending helper.""" 

2 

3import email.message 

4import email.policy 

5import email.utils 

6import smtplib 

7import ssl 

8 

9import gws 

10 

11gws.ext.new.helper('email') 

12 

13 

14class SmtpMode(gws.Enum): 

15 """SMTP connection modes.""" 

16 plain = 'plain' 

17 """Plain connection without encryption.""" 

18 ssl = 'ssl' 

19 """SSL connection.""" 

20 tls = 'tls' 

21 """TLS connection (STARTTLS).""" 

22 

23 

24class SmtpConfig(gws.Config): 

25 """SMTP server configuration.""" 

26 

27 mode: SmtpMode = SmtpMode.ssl 

28 """Connection mode.""" 

29 host: str 

30 """SMTP host name""" 

31 port: int = 0 

32 """SMTP port.""" 

33 login: str = '' 

34 """Login""" 

35 password: str = '' 

36 """Password.""" 

37 timeout: gws.Duration = '30' 

38 """Connection timeout.""" 

39 

40 

41class Config(gws.Config): 

42 """Mail helper settings""" 

43 

44 smtp: SmtpConfig 

45 """SMTP server configuration.""" 

46 mailFrom: str = '' 

47 """Default 'From' address.""" 

48 

49 

50class Message(gws.Data): 

51 """Email message.""" 

52 

53 subject: str 

54 """Subject.""" 

55 mailTo: str 

56 """To addresses, comma separated.""" 

57 mailFrom: str 

58 """From address (default if omitted).""" 

59 bcc: str 

60 """Bcc addresses.""" 

61 text: str 

62 """Plain text content.""" 

63 html: str 

64 """HTML content.""" 

65 

66 

67class Error(gws.Error): 

68 pass 

69 

70 

71## 

72 

73_DEFAULT_POLICY = { 

74 'linesep': '\r\n', 

75 'cte_type': '7bit', 

76 'utf8': False, 

77} 

78 

79_DEFAULT_ENCODING = 'quoted-printable' 

80 

81_DEFAULT_PORT = { 

82 SmtpMode.plain: 25, 

83 SmtpMode.ssl: 465, 

84 SmtpMode.tls: 587, 

85} 

86 

87 

88class _SmtpServer(gws.Data): 

89 mode: SmtpMode 

90 host: str 

91 port: int 

92 login: str 

93 password: str 

94 timeout: int 

95 

96 

97class Object(gws.Node): 

98 smtp: _SmtpServer 

99 mailFrom: str 

100 

101 def configure(self): 

102 self.mailFrom = self.cfg('mailFrom') 

103 

104 p = self.cfg('smtp') 

105 

106 self.smtp = _SmtpServer( 

107 mode=p.mode or SmtpMode.ssl, 

108 host=p.host, 

109 login=p.login, 

110 password=p.password, 

111 timeout=p.timeout, 

112 ) 

113 self.smtp.port = p.port or _DEFAULT_PORT.get(self.smtp.mode) 

114 

115 def send_mail(self, m: Message): 

116 msg = email.message.EmailMessage(email.policy.EmailPolicy(**_DEFAULT_POLICY)) 

117 

118 msg['Subject'] = m.subject 

119 msg['To'] = m.mailTo 

120 msg['From'] = m.mailFrom or self.mailFrom 

121 msg['Date'] = email.utils.formatdate() 

122 if m.bcc: 

123 msg['Bcc'] = m.bcc 

124 

125 msg.set_content(m.text, cte=_DEFAULT_ENCODING) 

126 if m.html: 

127 # @TODO images 

128 msg.add_alternative(m.html, subtype='html', cte=_DEFAULT_ENCODING) 

129 

130 self._send(msg) 

131 

132 def _send(self, msg): 

133 if self.smtp: 

134 try: 

135 with self._smtp_connection() as conn: 

136 conn.send_message(msg) 

137 except OSError as exc: 

138 raise Error('SMTP error') from exc 

139 

140 def _smtp_connection(self): 

141 if self.smtp.mode == SmtpMode.ssl: 

142 conn = smtplib.SMTP_SSL( 

143 host=self.smtp.host, 

144 port=self.smtp.port, 

145 timeout=self.smtp.timeout, 

146 context=ssl.create_default_context(), 

147 ) 

148 else: 

149 conn = smtplib.SMTP( 

150 host=self.smtp.host, 

151 port=self.smtp.port, 

152 timeout=self.smtp.timeout, 

153 ) 

154 

155 # conn.set_debuglevel(2) 

156 

157 if self.smtp.mode == SmtpMode.tls: 

158 conn.starttls(context=ssl.create_default_context()) 

159 

160 if self.smtp.login: 

161 conn.login(self.smtp.login, self.smtp.password) 

162 

163 return conn