Coverage for cc_modules/cc_email.py: 40%
99 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 14:23 +0100
1"""
2camcops_server/cc_modules/cc_email.py
4===============================================================================
6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CamCOPS.
11 CamCOPS is free software: you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation, either version 3 of the License, or
14 (at your option) any later version.
16 CamCOPS is distributed in the hope that it will be useful,
17 but WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 GNU General Public License for more details.
21 You should have received a copy of the GNU General Public License
22 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26**Email functions/log class.**
28"""
30import datetime
31import email.utils
32import logging
33from typing import Optional, Sequence, Tuple
35from cardinal_pythonlib.datetimefunc import (
36 convert_datetime_to_utc,
37 get_now_localtz_pendulum,
38 get_now_utc_pendulum,
39)
40from cardinal_pythonlib.email.sendmail import (
41 COMMASPACE,
42 make_email,
43 send_msg,
44 STANDARD_SMTP_PORT,
45 STANDARD_TLS_PORT,
46)
47from cardinal_pythonlib.httpconst import MimeType
48from cardinal_pythonlib.logs import BraceStyleAdapter
49from sqlalchemy.orm import Mapped, mapped_column, reconstructor
50from sqlalchemy.sql.sqltypes import (
51 BigInteger,
52 Integer,
53 Text,
54)
56from camcops_server.cc_modules.cc_sqlalchemy import Base
57from camcops_server.cc_modules.cc_sqla_coltypes import (
58 CharsetColType,
59 EmailAddressColType,
60 HostnameColType,
61 LongText,
62 MimeTypeColType,
63 Rfc2822DateColType,
64 UserNameExternalColType,
65)
67log = BraceStyleAdapter(logging.getLogger(__name__))
70# =============================================================================
71# Email class
72# =============================================================================
75class Email(Base):
76 """
77 Class representing an e-mail sent from CamCOPS.
79 This is abstract, in that it doesn't care about the purpose of the e-mail.
80 It's cross-referenced from classes that use it, such as
81 :class:`camcops_server.cc_modules.cc_exportmodels.ExportedTaskEmail`.
82 """
84 __tablename__ = "_emails"
86 # -------------------------------------------------------------------------
87 # Basic things
88 # -------------------------------------------------------------------------
89 id: Mapped[int] = mapped_column(
90 # SQLite doesn't support autoincrement with BigInteger
91 BigInteger().with_variant(Integer, "sqlite"),
92 primary_key=True,
93 autoincrement=True,
94 comment="Arbitrary primary key",
95 )
96 created_at_utc: Mapped[Optional[datetime.datetime]] = mapped_column(
97 comment="Date/time message was created (UTC)",
98 )
99 # -------------------------------------------------------------------------
100 # Headers
101 # -------------------------------------------------------------------------
102 date: Mapped[Optional[str]] = mapped_column(
103 Rfc2822DateColType, comment="Email date in RFC 2822 format"
104 )
105 from_addr: Mapped[Optional[str]] = mapped_column(
106 EmailAddressColType, comment="Email 'From:' field"
107 )
108 sender: Mapped[Optional[str]] = mapped_column(
109 EmailAddressColType, comment="Email 'Sender:' field"
110 )
111 reply_to: Mapped[Optional[str]] = mapped_column(
112 EmailAddressColType, comment="Email 'Reply-To:' field"
113 )
114 to: Mapped[Optional[str]] = mapped_column(
115 Text, comment="Email 'To:' field"
116 )
117 cc: Mapped[Optional[str]] = mapped_column(
118 Text, comment="Email 'Cc:' field"
119 )
120 bcc: Mapped[Optional[str]] = mapped_column(
121 Text, comment="Email 'Bcc:' field"
122 )
123 subject: Mapped[Optional[str]] = mapped_column(
124 Text, comment="Email 'Subject:' field"
125 )
126 # -------------------------------------------------------------------------
127 # Body, message
128 # -------------------------------------------------------------------------
129 body: Mapped[Optional[str]] = mapped_column(Text, comment="Email body")
130 content_type: Mapped[Optional[str]] = mapped_column(
131 MimeTypeColType, comment="MIME type for e-mail body"
132 )
133 charset: Mapped[Optional[str]] = mapped_column(
134 CharsetColType, comment="Character set for e-mail body"
135 )
136 msg_string: Mapped[Optional[str]] = mapped_column(
137 LongText, comment="Full encoded e-mail"
138 )
139 # -------------------------------------------------------------------------
140 # Server
141 # -------------------------------------------------------------------------
142 host: Mapped[Optional[str]] = mapped_column(
143 HostnameColType, comment="Email server"
144 )
145 port: Mapped[Optional[int]] = mapped_column(
146 comment="Port number on e-mail server"
147 )
148 username: Mapped[Optional[str]] = mapped_column(
149 UserNameExternalColType,
150 comment="Username on e-mail server",
151 )
152 use_tls: Mapped[Optional[bool]] = mapped_column(comment="Use TLS?")
153 # -------------------------------------------------------------------------
154 # Status
155 # -------------------------------------------------------------------------
156 sent: Mapped[bool] = mapped_column(default=False, comment="Sent?")
157 sent_at_utc: Mapped[Optional[datetime.datetime]] = mapped_column(
158 comment="Date/time message was sent (UTC)"
159 )
160 sending_failure_reason: Mapped[Optional[str]] = mapped_column(
161 Text, comment="Reason for sending failure"
162 )
164 def __init__(
165 self,
166 from_addr: str = "",
167 date: str = None,
168 sender: str = "",
169 reply_to: str = "",
170 to: str = "",
171 cc: str = "",
172 bcc: str = "",
173 subject: str = "",
174 body: str = "",
175 content_type: str = MimeType.TEXT,
176 charset: str = "utf8",
177 attachment_filenames: Sequence[str] = None,
178 attachments_binary: Sequence[Tuple[str, bytes]] = None,
179 save_msg_string: bool = False,
180 ) -> None:
181 """
182 Args:
183 from_addr: name of the sender for the "From:" field
184 date: e-mail date in RFC 2822 format, or ``None`` for "now"
185 sender: name of the sender for the "Sender:" field
186 reply_to: name of the sender for the "Reply-To:" field
188 to: e-mail address(es) of the recipients for "To:" field, as a
189 CSV list
190 cc: e-mail address(es) of the recipients for "Cc:" field, as a
191 CSV list
192 bcc: e-mail address(es) of the recipients for "Bcc:" field, as a
193 CSV list
195 subject: e-mail subject
196 body: e-mail body
197 content_type: MIME type for body content, default ``text/plain``
198 charset: character set for body; default ``utf8``
199 charset:
201 attachment_filenames: filenames of attachments to add
202 attachments_binary: binary attachments to add, as a list of
203 ``filename, bytes`` tuples
205 save_msg_string: save the encoded message string? (May take
206 significant space in the database).
207 """
208 # Note: we permit from_addr to be blank only for automated database
209 # copying.
211 # ---------------------------------------------------------------------
212 # Timestamp
213 # ---------------------------------------------------------------------
214 now_local = get_now_localtz_pendulum()
215 self.created_at_utc = convert_datetime_to_utc(now_local)
217 # -------------------------------------------------------------------------
218 # Arguments
219 # -------------------------------------------------------------------------
220 if not date:
221 date = email.utils.format_datetime(now_local)
222 attachment_filenames = attachment_filenames or []
223 attachments_binary = attachments_binary or []
224 if attachments_binary:
225 attachment_binary_filenames, attachment_binaries = zip(
226 *attachments_binary
227 )
228 else:
229 attachment_binary_filenames = [] # type: ignore[assignment] # type: ignore[no-redef] # noqa: E501
230 attachment_binaries = [] # type: ignore[assignment] # type: ignore[no-redef] # noqa: E501
231 # ... https://stackoverflow.com/questions/13635032/what-is-the-inverse-function-of-zip-in-python # noqa
232 # Other checks performed by our e-mail function below
234 # ---------------------------------------------------------------------
235 # Transient fields
236 # ---------------------------------------------------------------------
237 self.password = None
238 self.msg = (
239 make_email(
240 from_addr=from_addr,
241 date=date,
242 sender=sender,
243 reply_to=reply_to,
244 to=to,
245 cc=cc,
246 bcc=bcc,
247 subject=subject,
248 body=body,
249 content_type=content_type,
250 attachment_filenames=attachment_filenames,
251 attachment_binaries=attachment_binaries,
252 attachment_binary_filenames=attachment_binary_filenames,
253 )
254 if from_addr
255 else None
256 )
258 # ---------------------------------------------------------------------
259 # Database fields
260 # ---------------------------------------------------------------------
261 self.date = date
262 self.from_addr = from_addr
263 self.sender = sender
264 self.reply_to = reply_to
265 self.to = to
266 self.cc = cc
267 self.bcc = bcc
268 self.subject = subject
269 self.body = body
270 self.content_type = content_type
271 self.charset = charset
272 if save_msg_string:
273 self.msg_string = self.msg.as_string()
275 @reconstructor
276 def init_on_load(self) -> None:
277 """
278 Called when SQLAlchemy recreates an object; see
279 https://docs.sqlalchemy.org/en/latest/orm/constructors.html.
280 """
281 self.password = None
282 self.msg = None
284 def send(
285 self,
286 host: str,
287 username: str,
288 password: str,
289 port: int = None,
290 use_tls: bool = True,
291 ) -> bool:
292 """
293 Sends message and returns success.
294 """
295 if port is None:
296 port = STANDARD_TLS_PORT if use_tls else STANDARD_SMTP_PORT
298 msg = None
299 msg_string = None
300 if self.msg:
301 msg = self.msg
302 elif self.msg_string:
303 msg_string = self.msg_string
304 else:
305 log.error("Can't send message; not present (not saved?)")
306 return False
308 # Password not always required (for insecure servers...)
310 if self.sent:
311 log.info("Resending message")
313 self.host = host
314 self.port = port
315 self.username = username
316 # don't save password
317 self.use_tls = use_tls
318 to_addrs = COMMASPACE.join(
319 x for x in (self.to, self.cc, self.bcc) if x
320 )
321 header_components = filter(
322 None,
323 (
324 f"To: {self.to}" if self.to else "",
325 f"Cc: {self.cc}" if self.cc else "",
326 f"Bcc: {self.bcc}" if self.bcc else "",
327 f"Subject: {self.subject}" if self.subject else "",
328 ),
329 )
330 log.info("Sending email -- {}", " -- ".join(header_components))
331 try:
332 send_msg(
333 from_addr=self.from_addr,
334 to_addrs=to_addrs,
335 host=host,
336 user=username,
337 password=password,
338 port=port,
339 use_tls=use_tls,
340 msg=msg,
341 msg_string=msg_string,
342 )
343 log.debug("... sent")
344 self.sent = True
345 self.sent_at_utc = get_now_utc_pendulum()
346 self.sending_failure_reason = None
348 return True
349 except RuntimeError as e:
350 log.error("Failed to send e-mail: {!s}", e)
351 if not self.sent:
352 self.sent = False
353 self.sending_failure_reason = str(e)
355 return False