/
email.py
executable file
·240 lines (205 loc) · 7.84 KB
/
email.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
"""Utility class to simplify sending emails."""
import logging
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from os.path import expanduser, join
from typing import Any, List, Optional, Union
from .loader import load_json, load_yaml
from .typehint import ListTuple
try:
from email_validator import validate_email
except ImportError:
validate_email = None
logger = logging.getLogger(__name__)
class EmailConfigurationError(Exception):
pass
class Email:
"""Emailer utility. Parameters in dictionary or file (eg. yaml below):
| connection_type: "ssl" ("ssl" for smtp ssl or "lmtp", otherwise basic smtp is assumed)
| host: "localhost"
| port: 123
| local_hostname: "mycomputer.fqdn.com"
| timeout: 3
| username: "user"
| password: "pass"
| sender: "a@b.com" (if not supplied username is used as sender)
Args:
**kwargs: See below
email_config_dict (dict): HDX configuration dictionary OR
email_config_json (str): Path to JSON HDX configuration OR
email_config_yaml (str): Path to YAML HDX configuration. Defaults to ~/hdx_email_configuration.yaml.
"""
default_email_config_yaml = join(
expanduser("~"), "hdx_email_configuration.yaml"
)
def __init__(self, **kwargs: Any) -> None:
email_config_found = False
email_config_dict = kwargs.get("email_config_dict", None)
if email_config_dict:
email_config_found = True
logger.info("Loading email configuration from dictionary")
email_config_json = kwargs.get("email_config_json", "")
if email_config_json:
if email_config_found:
raise EmailConfigurationError(
"More than one email configuration file given!"
)
email_config_found = True
logger.info(
f"Loading email configuration from: {email_config_json}"
)
email_config_dict = load_json(email_config_json)
email_config_yaml = kwargs.get("email_config_yaml", None)
if email_config_found:
if email_config_yaml:
raise EmailConfigurationError(
"More than one email configuration file given!"
)
else:
if not email_config_yaml:
logger.info(
"No email configuration parameter. Using default email configuration path."
)
email_config_yaml = Email.default_email_config_yaml
logger.info(
f"Loading email configuration from: {email_config_yaml}"
)
email_config_dict = load_yaml(email_config_yaml)
self.connection_type = email_config_dict.get("connection_type", "smtp")
self.host = email_config_dict.get("host", "")
self.port = email_config_dict.get("port", 0)
self.local_hostname = email_config_dict.get("local_hostname")
self.timeout = email_config_dict.get("timeout")
self.source_address = email_config_dict.get("source_address")
self.username = email_config_dict.get("username")
self.password = email_config_dict.get("password")
self.sender = email_config_dict.get("sender", self.username)
self.server = None
def __enter__(self) -> "Email":
"""Return Email object for with statement.
Returns:
None
"""
return self
def __exit__(self, *args: Any) -> None:
"""Close Email object for end of with statement.
Args:
*args: Not used
Returns:
None
"""
def connect(self) -> None:
"""Connect to server.
Returns:
None
"""
if self.connection_type.lower() == "ssl":
self.server = smtplib.SMTP_SSL(
host=self.host,
port=self.port,
local_hostname=self.local_hostname,
timeout=self.timeout,
source_address=self.source_address,
)
elif self.connection_type.lower() == "lmtp":
self.server = smtplib.LMTP(
host=self.host,
port=self.port,
local_hostname=self.local_hostname,
source_address=self.source_address,
)
else:
self.server = smtplib.SMTP(
host=self.host,
port=self.port,
local_hostname=self.local_hostname,
timeout=self.timeout,
source_address=self.source_address,
)
self.server.login(self.username, self.password)
def close(self) -> None:
"""Close connection to email server.
Returns:
None
"""
self.server.quit()
@staticmethod
def get_normalised_emails(
recipients: Union[str, ListTuple[str]],
) -> List[str]:
"""Get list of normalised emails.
Args:
recipients (Union[str, ListTuple[str]]): Email recipient(s)
Returns:
List[str]: Normalised emails
"""
if isinstance(recipients, str):
recipients = (recipients,)
if validate_email is None:
return recipients
normalised_recipients = []
for recipient in recipients:
v = validate_email(
recipient, check_deliverability=True
) # validate and get info
normalised_recipients.append(
v.normalized
) # replace with normalized form
return normalised_recipients
def send(
self,
to: Union[str, ListTuple[str]],
subject: str,
text_body: str,
html_body: Optional[str] = None,
sender: Optional[str] = None,
cc: Union[str, ListTuple[str], None] = None,
bcc: Union[str, ListTuple[str], None] = None,
**kwargs: Any,
) -> None:
"""Send email. to, cc and bcc take either a string email address or a
list of string email addresses. cc and bcc default to None.
Args:
to (Union[str, ListTuple[str]]): Email recipient(s)
subject (str): Email subject
text_body (str): Plain text email body
html_body (Optional[str]): HTML email body
sender (Optional[str]): Email sender. Defaults to global sender.
cc (Union[str, ListTuple[str], None]): Email cc. Defaults to None.
bcc (Union[str, ListTuple[str], None]): Email bcc. Defaults to None.
**kwargs: See below
mail_options (List): Mail options (see smtplib documentation)
rcpt_options (List): Recipient options (see smtplib documentation)
Returns:
None
"""
if sender is None:
sender = self.sender
v = validate_email(
sender, check_deliverability=False
) # validate and get info
sender = v.normalized
if html_body is not None:
msg = MIMEMultipart("alternative")
part1 = MIMEText(text_body, "plain")
part2 = MIMEText(html_body, "html")
msg.attach(part1)
msg.attach(part2)
else:
msg = MIMEText(text_body)
msg["Subject"] = subject
msg["From"] = sender
normalised_to = self.get_normalised_emails(to)
msg["To"] = ", ".join(normalised_to)
if cc is not None:
normalised_cc = self.get_normalised_emails(cc)
msg["Cc"] = ", ".join(normalised_cc)
normalised_to.extend(normalised_cc)
if bcc is not None:
normalised_bcc = self.get_normalised_emails(bcc)
normalised_to.extend(normalised_bcc)
# Perform operations via server
self.connect()
self.server.sendmail(sender, normalised_to, msg.as_string(), **kwargs)
self.close()