django-anymail/anymail/backends/resend.py

import mimetypes
from email.charset import QP, Charset
from email.header import decode_header, make_header
from email.headerregistry import Address

from ..message import AnymailRecipientStatus
from ..utils import (
    BASIC_NUMERIC_TYPES,
    CaseInsensitiveCasePreservingDict,
    get_anymail_setting,
)
from .base_requests import AnymailRequestsBackend, RequestsPayload

# Used to force RFC-2047 encoded word
# in address formatting workaround
QP_CHARSET = Charset("utf-8")
QP_CHARSET.header_encoding = QP


class EmailBackend(AnymailRequestsBackend):
    """
    Resend (resend.com) API Email Backend
    """

    esp_name = "Resend"

    def __init__(self, **kwargs):
        """Init options from Django settings"""
        esp_name = self.esp_name
        self.api_key = get_anymail_setting(
            "api_key", esp_name=esp_name, kwargs=kwargs, allow_bare=True
        )
        api_url = get_anymail_setting(
            "api_url",
            esp_name=esp_name,
            kwargs=kwargs,
            default="https://api.resend.com/",
        )
        if not api_url.endswith("/"):
            api_url += "/"

        # Undocumented setting to control workarounds for Resend display-name issues
        # (see below). If/when Resend improves their API, you can disable Anymail's
        # workarounds by adding `"RESEND_WORKAROUND_DISPLAY_NAME_BUGS": False`
        # to your `ANYMAIL` settings.
        self.workaround_display_name_bugs = get_anymail_setting(
            "workaround_display_name_bugs",
            esp_name=esp_name,
            kwargs=kwargs,
            default=True,
        )

        super().__init__(api_url, **kwargs)

    def build_message_payload(self, message, defaults):
        return ResendPayload(message, defaults, self)

    def parse_recipient_status(self, response, payload, message):
        # Resend provides single message id, no other information.
        # Assume "queued".
        parsed_response = self.deserialize_json_response(response, payload, message)
        message_id = parsed_response["id"]
        recipient_status = CaseInsensitiveCasePreservingDict(
            {
                recip.addr_spec: AnymailRecipientStatus(
                    message_id=message_id, status="queued"
                )
                for recip in payload.recipients
            }
        )
        return dict(recipient_status)


class ResendPayload(RequestsPayload):
    def __init__(self, message, defaults, backend, *args, **kwargs):
        self.recipients = []  # for parse_recipient_status
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = "Bearer %s" % backend.api_key
        headers["Content-Type"] = "application/json"
        headers["Accept"] = "application/json"
        super().__init__(message, defaults, backend, headers=headers, *args, **kwargs)

    def get_api_endpoint(self):
        return "emails"

    def serialize_data(self):
        return self.serialize_json(self.data)

    #
    # Payload construction
    #

    def init_payload(self):
        self.data = {}  # becomes json

    def _resend_email_address(self, address):
        """
        Return EmailAddress address formatted for use with Resend.

        Works around a Resend bug that rejects properly formatted RFC 5322
        addresses that have the display-name enclosed in double quotes (e.g.,
        any display-name containing a comma), by substituting an RFC 2047
        encoded word.

        This works for all Resend address fields _except_ `from` (see below).
        """
        formatted = address.address
        if self.backend.workaround_display_name_bugs:
            if formatted.startswith('"'):
                # Workaround: force RFC-2047 encoded word
                formatted = str(
                    Address(
                        display_name=QP_CHARSET.header_encode(address.display_name),
                        addr_spec=address.addr_spec,
                    )
                )
        return formatted

    def set_from_email(self, email):
        # Can't use the address header workaround above for the `from` field:
        #   self.data["from"] = self._resend_email_address(email)
        # When `from` uses RFC-2047 encoding, Resend returns a "security_error"
        # status 451, "The email payload contain invalid characters".
        formatted = email.address
        if self.backend.workaround_display_name_bugs:
            if formatted.startswith("=?"):
                # Workaround: use an *unencoded* (Unicode str) display-name.
                # This allows use of non-ASCII characters (which Resend rejects when
                # encoded with RFC 2047). Some punctuation will still result in unusual
                # behavior or cause an "invalid `from` field" 422 error, but there's
                # nothing we can do about that.
                formatted = str(
                    # email.headerregistry.Address str format uses unencoded Unicode
                    Address(
                        # Convert RFC 2047 display name back to Unicode str
                        display_name=str(
                            make_header(decode_header(email.display_name))
                        ),
                        addr_spec=email.addr_spec,
                    )
                )
        self.data["from"] = formatted

    def set_recipients(self, recipient_type, emails):
        assert recipient_type in ["to", "cc", "bcc"]
        if emails:
            field = recipient_type
            self.data[field] = [self._resend_email_address(email) for email in emails]
            self.recipients += emails

    def set_subject(self, subject):
        self.data["subject"] = subject

    def set_reply_to(self, emails):
        if emails:
            self.data["reply_to"] = [
                self._resend_email_address(email) for email in emails
            ]

    def set_extra_headers(self, headers):
        # Resend requires header values to be strings (not integers) as of 2023-10-20.
        # Stringify ints and floats; anything else is the caller's responsibility.
        self.data.setdefault("headers", {}).update(
            {
                k: str(v) if isinstance(v, BASIC_NUMERIC_TYPES) else v
                for k, v in headers.items()
            }
        )

    def set_text_body(self, body):
        self.data["text"] = body

    def set_html_body(self, body):
        if "html" in self.data:
            # second html body could show up through multiple alternatives,
            # or html body + alternative
            self.unsupported_feature("multiple html parts")
        self.data["html"] = body

    @staticmethod
    def make_attachment(attachment):
        """Returns Resend attachment dict for attachment"""
        filename = attachment.name or ""
        if not filename:
            # Provide default name with reasonable extension.
            # (Resend guesses content type from the filename extension;
            # there doesn't seem to be any other way to specify it.)
            ext = mimetypes.guess_extension(attachment.content_type)
            if ext is not None:
                filename = f"attachment{ext}"
        att = {"content": attachment.b64content, "filename": filename}
        # attachment.inline / attachment.cid not supported
        return att

    def set_attachments(self, attachments):
        if attachments:
            if any(att.content_id for att in attachments):
                self.unsupported_feature("inline content-id")
            self.data["attachments"] = [
                self.make_attachment(attachment) for attachment in attachments
            ]

    def set_metadata(self, metadata):
        # Send metadata as json in a custom X-Metadata header.
        # (Resend's own "tags" are severely limited in character set)
        self.data.setdefault("headers", {})["X-Metadata"] = self.serialize_json(
            metadata
        )

    # Resend doesn't support delayed sending
    # def set_send_at(self, send_at):

    def set_tags(self, tags):
        # Send tags using a custom X-Tags header.
        # (Resend's own "tags" are severely limited in character set)
        self.data.setdefault("headers", {})["X-Tags"] = self.serialize_json(tags)

    # Resend doesn't support changing click/open tracking per message
    # def set_track_clicks(self, track_clicks):
    # def set_track_opens(self, track_opens):

    # Resend doesn't support server-rendered templates.
    # (Their template feature is rendered client-side,
    # using React in node.js.)
    # def set_template_id(self, template_id):
    # def set_merge_data(self, merge_data):
    # def set_merge_global_data(self, merge_global_data):
    # def set_merge_metadata(self, merge_metadata):

    def set_esp_extra(self, extra):
        self.data.update(extra)