115 multipart html and plain text emails (#179)

* implemented alternative_body

* added tests for alternative_body

* import root_validator

* added documentation in the MailMsg class

* Some mail clients only see both bodies with related

---------

Co-authored-by: Maximilian Leiwig <[email protected]>

Author: Synrom

fastapi_mail/msg.py

@@ -7,6 +7,7 @@
 from email.mime.text import MIMEText
 from email.utils import formatdate, make_msgid
 from typing import Any, Union
+from .schemas import MessageType, MultipartSubtypeEnum
 
 PY3 = sys.version_info[0] == 3
 
@@ -18,6 +19,7 @@ class MailMsg:
     :param: subject: Email subject header
     :param: recipients: List of email addresses
     :param: body: Plain text message or HTML message
+    :param: alternative_body: Plain text message or HTML message
     :param: template_body: Data to pass into chosen Jinja2 template
     :param: subtype: MessageType class. Type of body parameter, either "plain" or "html"
     :param: sender: Email sender address
@@ -36,6 +38,7 @@ def __init__(self, entries) -> None:
         self.attachments = entries.attachments
         self.subject = entries.subject
         self.body = entries.body
+        self.alternative_body = entries.alternative_body
         self.template_body = entries.template_body
         self.cc = entries.cc
         self.bcc = entries.bcc
@@ -88,13 +91,40 @@ async def attach_file(self, message: MIMEMultipart, attachment: Any):
 
             self.message.attach(part)
 
+    def attach_alternative(self, message: MIMEMultipart) -> MIMEMultipart:
+        """
+        Attaches an alternative body to a given message
+        """
+        tmpmsg = message
+        if self.subtype == MessageType.plain:
+            flipped_subtype = "html"
+        else:
+            flipped_subtype = "plain"
+        tmpmsg.attach(self._mimetext(self.alternative_body, flipped_subtype))
+        message = MIMEMultipart(MultipartSubtypeEnum.related.value)
+        message.set_charset(self.charset)
+        message.attach(tmpmsg)
+        return message
+
     async def _message(self, sender: str = None) -> Union[EmailMessage, Message]:
         """
         Creates the email message
         """
 
         self.message = MIMEMultipart(self.multipart_subtype.value)
         self.message.set_charset(self.charset)
+
+        if self.template_body:
+            self.message.attach(self._mimetext(self.template_body, self.subtype.value))
+        elif self.body:
+            self.message.attach(self._mimetext(self.body, self.subtype.value))
+
+        if (
+            self.alternative_body is not None
+            and self.multipart_subtype == MultipartSubtypeEnum.alternative
+        ):
+            self.message = self.attach_alternative(self.message)
+
         self.message["Date"] = formatdate(time.time(), localtime=True)
         self.message["Message-ID"] = self.msgId
         self.message["To"] = ", ".join(self.recipients)
@@ -112,12 +142,6 @@ async def _message(self, sender: str = None) -> Union[EmailMessage, Message]:
         if self.reply_to:
             self.message["Reply-To"] = ", ".join(self.reply_to)
 
-        if self.template_body:
-            self.message.attach(self._mimetext(self.template_body, self.subtype.value))
-
-        elif self.body:
-            self.message.attach(self._mimetext(self.body, self.subtype.value))
-
         if self.attachments:
             await self.attach_file(self.message, self.attachments)
 

fastapi_mail/schemas.py

@@ -4,7 +4,7 @@
 from mimetypes import MimeTypes
 from typing import Dict, List, Optional, Union
 
-from pydantic import BaseModel, EmailStr, validator
+from pydantic import BaseModel, EmailStr, validator, root_validator
 from starlette.datastructures import Headers, UploadFile
 
 from fastapi_mail.errors import WrongFile
@@ -38,6 +38,7 @@ class MessageSchema(BaseModel):
     attachments: List[Union[UploadFile, Dict, str]] = []
     subject: str = ""
     body: Optional[Union[str, list]] = None
+    alternative_body: Optional[str] = None
     template_body: Optional[Union[list, dict, str]] = None
     cc: List[EmailStr] = []
     bcc: List[EmailStr] = []
@@ -94,5 +95,17 @@ def validate_subtype(cls, value, values, config, field):
             return MessageType.html
         return value
 
+    @root_validator
+    def validate_alternative_body(cls, values):
+        """
+        Validate alternative_body field
+        """
+        if (
+            values["multipart_subtype"] != MultipartSubtypeEnum.alternative
+            and values["alternative_body"]
+        ):
+            values["alternative_body"] = None
+        return values
+
     class Config:
         arbitrary_types_allowed = True

tests/test_connection.py

@@ -2,7 +2,13 @@
 
 import pytest
 
-from fastapi_mail import ConnectionConfig, FastMail, MessageSchema, MessageType
+from fastapi_mail import (
+    ConnectionConfig,
+    FastMail,
+    MessageSchema,
+    MessageType,
+    MultipartSubtypeEnum,
+)
 
 CONTENT = "This file contents some information."
 
@@ -81,10 +87,7 @@ async def test_attachement_message(mail_config):
 
         assert len(outbox) == 1
         assert mail._payload[1].get_content_maintype() == "application"
-        assert (
-            mail._payload[1].__dict__.get("_headers")[0][1]
-            == "application/octet-stream"
-        )
+        assert mail._payload[1].__dict__.get("_headers")[0][1] == "application/octet-stream"
 
 
 @pytest.mark.asyncio
@@ -146,20 +149,16 @@ async def test_attachement_message_with_headers(mail_config):
 
         assert len(outbox) == 1
         mail = outbox[0]
-        assert mail._payload[1].get_content_maintype() == msg.attachments[0][1].get(
-            "mime_type"
-        )
-        assert mail._payload[1].get_content_subtype() == msg.attachments[0][1].get(
-            "mime_subtype"
-        )
+        assert mail._payload[1].get_content_maintype() == msg.attachments[0][1].get("mime_type")
+        assert mail._payload[1].get_content_subtype() == msg.attachments[0][1].get("mime_subtype")
 
         assert mail._payload[1].__dict__.get("_headers")[0][1] == "image/png"
-        assert mail._payload[1].__dict__.get("_headers")[3][1] == msg.attachments[0][
-            1
-        ].get("headers").get("Content-ID")
-        assert mail._payload[1].__dict__.get("_headers")[4][1] == msg.attachments[0][
-            1
-        ].get("headers").get("Content-Disposition")
+        assert mail._payload[1].__dict__.get("_headers")[3][1] == msg.attachments[0][1].get(
+            "headers"
+        ).get("Content-ID")
+        assert mail._payload[1].__dict__.get("_headers")[4][1] == msg.attachments[0][1].get(
+            "headers"
+        ).get("Content-Disposition")
 
         assert (
             mail._payload[2].__dict__.get("_headers")[3][1] == "attachment; "
@@ -183,9 +182,7 @@ async def test_jinja_message_list(mail_config):
     fm = FastMail(conf)
 
     with fm.record_messages() as outbox:
-        await fm.send_message(
-            message=msg, template_name="array_iteration_jinja_template.html"
-        )
+        await fm.send_message(message=msg, template_name="array_iteration_jinja_template.html")
 
         assert len(outbox) == 1
         mail = outbox[0]
@@ -284,10 +281,84 @@ async def test_jinja_message_with_html(mail_config):
     )
     conf = ConnectionConfig(**mail_config)
     fm = FastMail(conf)
-    await fm.send_message(
-        message=msg, template_name="array_iteration_jinja_template.html"
-    )
+    await fm.send_message(message=msg, template_name="array_iteration_jinja_template.html")
 
     assert msg.template_body == ("\n    \n    \n        Andrej\n    \n\n")
 
     assert not msg.body
+
+
+@pytest.mark.asyncio
+async def test_send_msg_with_alternative_body(mail_config):
+    msg = MessageSchema(
+        subject="testing",
+        recipients=["[email protected]"],
+        body="<p Test data </p>",
+        subtype=MessageType.html,
+        alternative_body="Test data",
+        multipart_subtype=MultipartSubtypeEnum.alternative,
+    )
+
+    sender = f"{mail_config['MAIL_FROM_NAME']} <{mail_config['MAIL_FROM']}>"
+    conf = ConnectionConfig(**mail_config)
+    fm = FastMail(conf)
+    fm.config.SUPPRESS_SEND = 1
+    with fm.record_messages() as outbox:
+        await fm.send_message(message=msg)
+
+        mail = outbox[0]
+        assert len(outbox) == 1
+        body = mail._payload[0]
+        assert len(body._payload) == 2
+        assert body._headers[1][1] == 'multipart/alternative; charset="utf-8"'
+        assert mail["subject"] == "testing"
+        assert mail["from"] == sender
+        assert mail["To"] == "[email protected]"
+
+        assert body._payload[0]._headers[0][1] == 'text/html; charset="utf-8"'
+        assert body._payload[1]._headers[0][1] == 'text/plain; charset="utf-8"'
+    assert msg.alternative_body == "Test data"
+    assert msg.multipart_subtype == MultipartSubtypeEnum.alternative
+
+
+@pytest.mark.asyncio
+async def test_send_msg_with_alternative_body_and_attachements(mail_config):
+    directory = os.getcwd()
+    text_file = directory + "/tests/txt_files/plain.txt"
+
+    with open(text_file, "w") as file:
+        file.write(CONTENT)
+
+    subject = "testing"
+    to = "[email protected]"
+    msg = MessageSchema(
+        subject=subject,
+        recipients=[to],
+        body="html test",
+        subtype="html",
+        attachments=[text_file],
+        alternative_body="plain test",
+        multipart_subtype="alternative",
+    )
+    sender = f"{mail_config['MAIL_FROM_NAME']} <{mail_config['MAIL_FROM']}>"
+    conf = ConnectionConfig(**mail_config)
+    fm = FastMail(conf)
+    fm.config.SUPPRESS_SEND = 1
+    with fm.record_messages() as outbox:
+        await fm.send_message(message=msg)
+
+        mail = outbox[0]
+
+        assert len(outbox) == 1
+        body = mail._payload[0]
+        assert len(body._payload) == 2
+        assert body._headers[1][1] == 'multipart/alternative; charset="utf-8"'
+        assert mail["subject"] == "testing"
+        assert mail["from"] == sender
+        assert mail["To"] == "[email protected]"
+
+        assert body._payload[0]._headers[0][1] == 'text/html; charset="utf-8"'
+        assert body._payload[1]._headers[0][1] == 'text/plain; charset="utf-8"'
+
+        assert mail._payload[1].get_content_maintype() == "application"
+        assert mail._payload[1].__dict__.get("_headers")[0][1] == "application/octet-stream"

tests/test_message.py

@@ -4,6 +4,7 @@
 
 from fastapi_mail.msg import MailMsg
 from fastapi_mail.schemas import MessageSchema, MessageType, MultipartSubtypeEnum
+from pydantic.error_wrappers import ValidationError
 
 
 def test_initialize():
@@ -188,3 +189,26 @@ async def test_message_charset():
     msg_object = await msg._message("[email protected]")
     assert msg_object._charset is not None
     assert msg_object._charset == "utf-8"
+
+
+def test_message_with_alternative_body_but_wrong_multipart_subtype():
+    message = MessageSchema(
+        subject="test subject",
+        recipients=["[email protected]"],
+        body="test",
+        subtype=MessageType.plain,
+        alternative_body="alternative",
+    )
+    assert message.alternative_body == None
+
+
+def test_message_with_alternative_body():
+    message = MessageSchema(
+        subject="test subject",
+        recipients=["[email protected]"],
+        body="test",
+        subtype=MessageType.plain,
+        multipart_subtype=MultipartSubtypeEnum.alternative,
+        alternative_body="alternative",
+    )
+    assert message.alternative_body == "alternative"