Python3 AWS Lambda Send email with CSV attachment full code

data_source.py

"""
Again, don't just copy and paste the code below.  This is for your study only.

Change the code to suit your case accordingly.
"""
import logging
import csv
import zipfile

from zoneinfo import ZoneInfo
from datetime import datetime

from campaign_scripts.config import connect_rds

logger = logging.getLogger()
logger.setLevel(logging.INFO)

class MyDataSource:

    def __init__(self):
        self.csv_file_name = f'MyDataSource.{datetime.now(tz=ZoneInfo("Asia/Hong_Kong")).strftime("%Y%m%d_%H%M%S")}.csv'
        self.full_csv_path = f'/tmp/{self.csv_file_name}'
        self.zip_file_name = f'/tmp/{self.csv_file_name}.zip'

    def get_recipients(self) -> list:
        return [
            '[email protected]'
        ]

    def get_email_subject(self) -> str:
        subject = f'[MyDataSource] Campaign Report {datetime.now(tz=ZoneInfo("Asia/Hong_Kong")).strftime("%Y-%m-%d %H:%M:%S")}'
        return subject

    def get_attachment_path(self) -> str:
        connection = connect_rds()

        with connection.cursor() as cursor:
            sql = "SELECT * FROM MyDataSource"
            cursor.execute( sql )
            result = cursor.fetchall()

        with open(self.full_csv_path, 'w', encoding='utf-8-sig') as csvfile:
            myFile = csv.writer(csvfile, dialect="excel")

            colname = self.get_colname()
            myFile.writerow(colname)

            for val in enumerate(result):
                myFile.writerow([
                    val[1]["id"],
                    val[1]["reqName"],
                    val[1]["reqPhone"],
                    val[1]["reqEmail"],
                    val[1]["reqVisitors"],
                    val[1]["reqPromo"],
                    val[1]["created_date"],
                ])

        # Create zip file from csv file
        with zipfile.ZipFile(self.zip_file_name, 'w') as zip:
            zip.write(self.full_csv_path)

        return self.zip_file_name

    def get_colname(self):
        return [
            "ID",
            "reqName",
            "reqPhone",
            "reqEmail",
            "reqVisitors",
            "reqPromo",
            "created_date",
        ]

lambda_function.py

"""
This file is for reference only.  Copy and paste it to your file will guarantee not working.

The purpose of this file is to show how to send AWS SES email via Python Lambda with email attachment.

Ref: https://stackoverflow.com/questions/77870505/python-3-12-write-chinese-in-excel-csv-utf-8-sig-not-work/
"""
import json
import os
import boto3
from botocore.exceptions import ClientError
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication

from campaign_scripts import get_campaigns

# References:
#
# https://aws.plainenglish.io/sending-emails-with-aws-lambda-aws-simple-email-service-ses-513839bc53ab
# https://docs.aws.amazon.com/ses/latest/dg/send-email-raw.html

def lambda_handler(event, context):

    campaigns = get_campaigns()

    for campaign in campaigns:
        # https://docs.aws.amazon.com/ses/latest/dg/send-email-raw.html

        # Replace [email protected] with your "From" address.
        # This address must be verified with Amazon SES.
        SENDER = "My Sender <[email protected]>"

        # Replace [email protected] with a "To" address. If your account
        # is still in the sandbox, this address must be verified.
        
        # Return a list of email
        RECIPIENT = campaign.get_recipients()

        # Specify a configuration set. If you do not want to use a configuration
        # set, comment the following variable, and the
        # ConfigurationSetName=CONFIGURATION_SET argument below.
        # CONFIGURATION_SET = "ConfigSet"

        # If necessary, replace us-west-2 with the AWS Region you're using for Amazon SES.
        AWS_REGION = "ap-southeast-1"

        # The subject line for the email. 
        # Return as string
        SUBJECT = campaign.get_email_subject()

        # The full path to the file that will be attached to the email.
        # Return a string.  This is the file path.
        ATTACHMENT = campaign.get_attachment_path()

        # The email body for recipients with non-HTML email clients.
        BODY_TEXT = "To whom it may concern,\r\n\r\nPlease find the attached file for your handling.\r\n\r\nHave a good day.\r\nBest Regards,\r\n\r\nSystem"

        # The HTML body of the email.
        BODY_HTML = """\
        <html><head></head><body>
        <p>To whom it may concern,</p>
        <p>Please find the attached file for your handling.</p>
        <p>Have a good good day.</p>
        <p>Best Regards,</p>
        <p></p>
        <p>System</p>
        </body></html>
        """

        # The character encoding for the email.
        CHARSET = "utf-8"

        # Create a new SES resource and specify a region.
        client = boto3.client('ses',region_name=AWS_REGION)

        # Create a multipart/mixed parent container.
        msg = MIMEMultipart('mixed')
        # Add subject, from and to lines.
        msg['Subject'] = SUBJECT
        msg['From'] = SENDER
        # msg['To'] = RECIPIENT

        # Create a multipart/alternative child container.
        msg_body = MIMEMultipart('alternative')

        # Encode the text and HTML content and set the character encoding. This step is
        # necessary if you're sending a message with characters outside the ASCII range.
        textpart = MIMEText(BODY_TEXT.encode(CHARSET), 'plain', CHARSET)
        htmlpart = MIMEText(BODY_HTML.encode(CHARSET), 'html', CHARSET)

        # Add the text and HTML parts to the child container.
        msg_body.attach(textpart)
        msg_body.attach(htmlpart)

        # Define the attachment part and encode it using MIMEApplication.
        # !!! Important: You are going to zip the CSV file and attach here, so it is `rb`, not just `r`
        att = MIMEApplication(open(ATTACHMENT, 'rb').read())

        # Add a header to tell the email client to treat this part as an attachment,
        # and to give the attachment a name.
        att.add_header('Content-Disposition','attachment',filename=os.path.basename(ATTACHMENT))

        # Attach the multipart/alternative child container to the multipart/mixed
        # parent container.
        msg.attach(msg_body)

        # Add the attachment to the parent container.
        msg.attach(att)

        response = ''

        try:
            #Provide the contents of the email.
            response = client.send_raw_email(
                Source=SENDER,
                Destinations=RECIPIENT,
                RawMessage={
                    'Data':msg.as_string(),
                }
            )
        # Display an error if something goes wrong.
        except ClientError as e:
            print(e.response['Error']['Message'])
        else:
            print("Email sent! Message ID:"),
            print(response['MessageId'])
            print(f'Attachment: {ATTACHMENT}')

    return {
        'statusCode': 200,
        'body': json.dumps('Email sent.'),
        'resp': response
    }