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 plain = 'plain' 

16 ssl = 'ssl' 

17 tls = 'tls' 

18 

19 

20class SmtpConfig(gws.Config): 

21 """SMTP server configuration. (added in 8.1)""" 

22 

23 mode: SmtpMode = 'ssl' 

24 """Connection mode.""" 

25 host: str 

26 """SMTP host name""" 

27 port: int = 0 

28 """SMTP port.""" 

29 login: str = '' 

30 """Login""" 

31 password: str = '' 

32 """Password.""" 

33 timeout: gws.Duration = 30 

34 """Connection timeout.""" 

35 

36 

37class Config(gws.Config): 

38 """Mail helper settings""" 

39 

40 smtp: SmtpConfig 

41 """SMTP server configuration.""" 

42 mailFrom: str = '' 

43 """Default 'From' address.""" 

44 

45 

46class Message(gws.Data): 

47 """Email message.""" 

48 

49 subject: str 

50 """Subject.""" 

51 mailTo: str 

52 """To addresses, comma separated.""" 

53 mailFrom: str 

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

55 bcc: str 

56 """Bcc addresses.""" 

57 text: str 

58 """Plain text content.""" 

59 html: str 

60 """HTML content.""" 

61 

62 

63class Error(gws.Error): 

64 pass 

65 

66 

67## 

68 

69_DEFAULT_POLICY = { 

70 'linesep': '\r\n', 

71 'cte_type': '7bit', 

72 'utf8': False, 

73} 

74 

75_DEFAULT_ENCODING = 'quoted-printable' 

76 

77_DEFAULT_PORT = { 

78 SmtpMode.plain: 25, 

79 SmtpMode.ssl: 465, 

80 SmtpMode.tls: 587, 

81} 

82 

83 

84class _SmtpServer(gws.Data): 

85 mode: SmtpMode 

86 host: str 

87 port: int 

88 login: str 

89 password: str 

90 timeout: int 

91 

92 

93class Object(gws.Node): 

94 smtp: _SmtpServer 

95 mailFrom: str 

96 

97 def configure(self): 

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

99 

100 p = self.cfg('smtp') 

101 

102 self.smtp = _SmtpServer( 

103 mode=p.mode or SmtpMode.ssl, 

104 host=p.host, 

105 login=p.login, 

106 password=p.password, 

107 timeout=p.timeout, 

108 ) 

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

110 

111 def send_mail(self, m: Message): 

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

113 

114 msg['Subject'] = m.subject 

115 msg['To'] = m.mailTo 

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

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

118 if m.bcc: 

119 msg['Bcc'] = m.bcc 

120 

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

122 if m.html: 

123 # @TODO images 

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

125 

126 self._send(msg) 

127 

128 def _send(self, msg): 

129 if self.smtp: 

130 try: 

131 with self._smtp_connection() as conn: 

132 conn.send_message(msg) 

133 except OSError as exc: 

134 raise Error('SMTP error') from exc 

135 

136 def _smtp_connection(self): 

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

138 conn = smtplib.SMTP_SSL( 

139 host=self.smtp.host, 

140 port=self.smtp.port, 

141 timeout=self.smtp.timeout, 

142 context=ssl.create_default_context(), 

143 ) 

144 else: 

145 conn = smtplib.SMTP( 

146 host=self.smtp.host, 

147 port=self.smtp.port, 

148 timeout=self.smtp.timeout, 

149 ) 

150 

151 # conn.set_debuglevel(2) 

152 

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

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

155 

156 if self.smtp.login: 

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

158 

159 return conn