django_amazon_ses.py
"""Boto3 email backend class for Amazon SES."""
import boto3
from botocore.exceptions import BotoCoreError, ClientError
from django.conf import settings
from django.core.mail.backends.base import BaseEmailBackend
from django.core.mail.message import sanitize_address
from django.dispatch import Signal
pre_send = Signal()
post_send = Signal()
class EmailBackend(BaseEmailBackend):
"""An email backend for use with Amazon SES.
Attributes:
conn: A client connection for Amazon SES.
"""
def __init__(
self,
fail_silently=False,
aws_access_key_id=None,
aws_secret_access_key=None,
**kwargs
):
"""Creates a client for the Amazon SES API.
Args:
fail_silently: Flag that determines whether Amazon SES
client errors should throw an exception.
"""
super().__init__(fail_silently=fail_silently)
# Get configuration from AWS prefixed settings in settings.py
access_key_id = getattr(settings, "AWS_ACCESS_KEY_ID", None)
secret_access_key = getattr(settings, "AWS_SECRET_ACCESS_KEY", None)
region_name = getattr(settings, "AWS_DEFAULT_REGION", "us-east-1")
# Override AWS prefixed configuration with Amazon SES-specific settings
access_key_id = getattr(settings, "AWS_SES_ACCESS_KEY_ID", access_key_id)
secret_access_key = getattr(
settings, "AWS_SES_SECRET_ACCESS_KEY", secret_access_key
)
region_name = getattr(settings, "AWS_SES_REGION", region_name)
self.configuration_set_name = getattr(
settings, "AWS_SES_CONFIGURATION_SET_NAME", None
)
self.tags = getattr(settings, "AWS_SES_TAGS", None)
# Override all previous configuration if settings provided
# through the constructor
if aws_access_key_id is not None and aws_secret_access_key is not None:
access_key_id = aws_access_key_id
secret_access_key = aws_secret_access_key
self.conn = boto3.client(
"ses",
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=region_name,
)
def send_messages(self, email_messages):
"""Sends one or more EmailMessage objects and returns the
number of email messages sent.
Args:
email_messages: A list of Django EmailMessage objects.
Returns:
An integer count of the messages sent.
Raises:
ClientError: An interaction with the Amazon SES HTTP API
failed.
"""
if not email_messages:
return 0
sent_message_count = 0
for email_message in email_messages:
if self._send(email_message):
sent_message_count += 1
return sent_message_count
def _send(self, email_message):
"""Sends an individual message via the Amazon SES HTTP API.
Args:
email_message: A single Django EmailMessage object.
Returns:
True if the EmailMessage was sent successfully, otherwise False.
Raises:
ClientError: An interaction with the Amazon SES HTTP API
failed.
"""
pre_send.send(self.__class__, message=email_message)
if not email_message.recipients():
return False
from_email = sanitize_address(email_message.from_email, email_message.encoding)
recipients = [
sanitize_address(addr, email_message.encoding)
for addr in email_message.recipients()
]
message = email_message.message().as_bytes(linesep="\r\n")
try:
kwargs = {
"Source": from_email,
"Destinations": recipients,
"RawMessage": {"Data": message},
}
if self.configuration_set_name is not None:
kwargs["ConfigurationSetName"] = self.configuration_set_name
if self.tags is not None:
kwargs["Tags"] = self.tags
result = self.conn.send_raw_email(**kwargs)
message_id = result["MessageId"]
post_send.send(self.__class__, message=email_message, message_id=message_id)
except (ClientError, BotoCoreError):
if not self.fail_silently:
raise
return False
return True