How to use ses_client method in localstack

Best Python code snippet using localstack_python

test_ses.py

Source:test_ses.py

1import json2import os3from datetime import date, datetime4import pytest5import localstack.config as config6TEST_TEMPLATE_ATTRIBUTES = {7    "TemplateName": "hello-world",8    "SubjectPart": "Subject test",9    "TextPart": "hello\nworld",10    "HtmlPart": "hello<br/>world",11}12@pytest.fixture13def create_template(ses_client):14    created_template_names = []15    def _create_template(Template):16        ses_client.create_template(Template=Template)17        created_template_names.append(Template["TemplateName"])18    yield _create_template19    for name in created_template_names:20        ses_client.delete_template(TemplateName=name)21class TestSES:22    def test_list_templates(self, ses_client, create_template):23        create_template(Template=TEST_TEMPLATE_ATTRIBUTES)24        templ_list = ses_client.list_templates()["TemplatesMetadata"]25        assert 1 == len(templ_list)26        created_template = templ_list[0]27        assert TEST_TEMPLATE_ATTRIBUTES["TemplateName"] == created_template["Name"]28        assert type(created_template["CreatedTimestamp"]) in (date, datetime)29        # Should not fail after 2 consecutive tries30        templ_list = ses_client.list_templates()["TemplatesMetadata"]31        assert 1 == len(templ_list)32        created_template = templ_list[0]33        assert TEST_TEMPLATE_ATTRIBUTES["TemplateName"] == created_template["Name"]34        assert type(created_template["CreatedTimestamp"]) in (date, datetime)35    def test_delete_template(self, ses_client, create_template):36        templ_list = ses_client.list_templates()["TemplatesMetadata"]37        assert 0 == len(templ_list)38        create_template(Template=TEST_TEMPLATE_ATTRIBUTES)39        templ_list = ses_client.list_templates()["TemplatesMetadata"]40        assert 1 == len(templ_list)41        ses_client.delete_template(TemplateName=TEST_TEMPLATE_ATTRIBUTES["TemplateName"])42        templ_list = ses_client.list_templates()["TemplatesMetadata"]43        assert 0 == len(templ_list)44    def test_get_identity_verification_attributes(self, ses_client):45        domain = "example.com"46        email = "user@example.com"47        test_values = [domain, email]48        response = ses_client.get_identity_verification_attributes(Identities=test_values)[49            "VerificationAttributes"50        ]51        assert 2 == len(response)52        for value in test_values:53            assert "Success" == response[value]["VerificationStatus"]54        assert "VerificationToken" in response[domain]55        assert "VerificationToken" not in response[email]56    def test_send_email_save(self, ses_client):57        data_dir = config.dirs.data or config.dirs.tmp58        email = "user@example.com"59        ses_client.verify_email_address(EmailAddress=email)60        message = ses_client.send_email(61            Source=email,62            Message={63                "Subject": {64                    "Data": "A_SUBJECT",65                },66                "Body": {67                    "Text": {68                        "Data": "A_MESSAGE",69                    },70                },71            },72            Destination={73                "ToAddresses": ["success@example.com"],74            },75        )76        with open(os.path.join(data_dir, "ses", message["MessageId"] + ".json"), "r") as f:77            message = f.read()78        contents = json.loads(message)79        assert email == contents["Source"]80        assert "A_SUBJECT" == contents["Subject"]81        assert "A_MESSAGE" == contents["Body"]82        assert ["success@example.com"] == contents["Destinations"]["ToAddresses"]83    def test_send_templated_email_save(self, ses_client, create_template):84        data_dir = config.dirs.data or config.dirs.tmp85        email = "user@example.com"86        ses_client.verify_email_address(EmailAddress=email)87        ses_client.delete_template(TemplateName=TEST_TEMPLATE_ATTRIBUTES["TemplateName"])88        create_template(Template=TEST_TEMPLATE_ATTRIBUTES)89        message = ses_client.send_templated_email(90            Source=email,91            Template=TEST_TEMPLATE_ATTRIBUTES["TemplateName"],92            TemplateData='{"A key": "A value"}',93            Destination={94                "ToAddresses": ["success@example.com"],95            },96        )97        with open(os.path.join(data_dir, "ses", message["MessageId"] + ".json"), "r") as f:98            message = f.read()99        contents = json.loads(message)100        assert email == contents["Source"]101        assert [TEST_TEMPLATE_ATTRIBUTES["TemplateName"]] == contents["Template"]102        assert ['{"A key": "A value"}'] == contents["TemplateData"]...

simple_email_service.py

Source:simple_email_service.py

1import boto32from email import encoders3from email.mime.base import MIMEBase4from email.mime.application import MIMEApplication5from email.mime.multipart import MIMEMultipart6from email.mime.text import MIMEText7import csv8import io91011class SES():12    def __init__(self) -> None:13        self.ses_client = boto3.client('ses')14    15    def verify_client(self, email_address:str):16        response = self.ses_client.verify_email_address(EmailAddress=email_address)17        return response18    19    def delete_client(self, email_address:str):20        response = self.ses_client.delete_identity(Identity=email_address)21        return response22    23    def send_email(self, email_id, to, data):24        msg = MIMEMultipart()25        msg["Subject"] = "Search Result of Cards"26        msg["From"] = email_id27        msg["To"] = to2829        html = """\30                <html>31                    <head></head>32                    <body>33                        <p>Hi!<br>34                        Download your search result file in this email.35                        </p>36                    </body>37                </html>38            """39        # Set message body40        body = MIMEText(html, "html")41        msg.attach(body)4243        output = io.StringIO()44        writer = csv.writer(output)45        for item in data:46            writer.writerow([item['CardID'], item['CardName'],47                             item['email'], item['contact1'], item['contact2'], 48                             item['website'], item['add']])49        50        output.seek(0)51        filename = "card_results.csv"52        part = MIMEApplication(output.read())53        part.add_header("Content-Disposition",54                        "attachment",55                        filename=filename)56        msg.attach(part)5758        response = self.ses_client.send_raw_email(59            Source=email_id,60            Destinations=[to],61            RawMessage={"Data": msg.as_string()}62        )63        return response
...

billing_mail.py

Source:billing_mail.py

1import os2import boto33import datetime4MAIL_FROM = os.environ['MAIL_FROM']5MAIL_TO   = os.environ['MAIL_TO']6cloud_watch_client = boto3.client('cloudwatch', region_name='us-east-1')7ses_client = boto3.client('ses', region_name='us-east-1')8def lambda_handler(event, context):9    if not is_registerd_email_address():10        register_email_address()11        return12    billing = getBilling()13    send_email(billing)14    return15def is_registerd_email_address():16    return len(ses_client.list_verified_email_addresses()["VerifiedEmailAddresses"]) == 117def register_email_address():18    return ses_client.verify_email_address(19        EmailAddress=MAIL_FROM20    )21def send_email(billing):22    return ses_client.send_email(23        Source=MAIL_FROM,24        Destination={25            'ToAddresses': [26                MAIL_TO,27            ]28        },29        Message={30            'Subject': {31                'Data': 'ãã·ã¹ãã ã¡ã¼ã«ãç¾å¨ã®AWSå©ç¨æé',32            },33            'Body': {34                'Text': {35                    'Data': 'ç¾å¨ã®AWSå©ç¨æã¯ã$' + str(billing) + ' ã§ãã' ,36                },37            }38        }39    )40def getBilling():41    start_time = datetime.datetime.today() - datetime.timedelta(days=1)42    end_time   = datetime.datetime.today()43    statistics = cloud_watch_client.get_metric_statistics(44                        Namespace='AWS/Billing',45                        MetricName='EstimatedCharges',46                        Dimensions=[47                            {48                                'Name': 'Currency',49                                'Value': 'USD'50                            }51                        ],52                        StartTime=start_time,53                        EndTime=end_time,54                        Period=3600*24,55                        Statistics=['Maximum']56                        )...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.