"""AWS utilities.""" from email.mime.multipart import MIMEMultipart from common.src import logger from common.src import aws from common.src.typings import Emails from . import config from .constants import SesPayload logger = logger.new_logger(__name__) def new_ses_client(): """SES client factory.""" return _new_aws_client(service_name=aws.AWSServices.SES) def new_s3_client(): """S3 client factory.""" return _new_aws_client(service_name=aws.AWSServices.S3) def _new_aws_client(**kwargs): """AWS client factory. KwArgs: Allowed any kwargs for boto3.client. """ return aws.Session().client(**kwargs, region_name=config.AWS_REGION) def send_raw_email(destinations: Emails, raw_message: MIMEMultipart): """High-level convenience function to send raw email via SES, with automatic SES client creation. Args: destinations: List of email addresses of the recipients (e.g. ["john.doe@theorchard.com", "jane.doe@theorchard.com"]) raw_message: MIME multipart message object. """ payload = { SesPayload.SOURCE: config.EMAIL_FROM, SesPayload.DESTINATIONS: destinations, SesPayload.RAW_MESSAGE: {SesPayload.DATA: raw_message.as_string()}, # Include SourceArn, FromArn and ReturnPathArn # or there will be permissions issues SesPayload.SOURCE_ARN: config.AWS_SES_SOURCE_ARN, SesPayload.FROM_ARN: config.AWS_SES_FROM_ARN, SesPayload.RETURN_PATH_ARN: config.AWS_SES_RETURN_PATH_ARN, } logger.debug( ">>> Sending email via SES: destinations=%s, source_arn=%s, from_arn=%s, " "return_path_arn=%s", destinations, config.AWS_SES_SOURCE_ARN, ) ses_client = new_ses_client() return ses_client.send_raw_email(**payload)