"""Handles incoming sesv2 requests, invokes methods, returns responses.""" import base64 import json from typing import List from urllib.parse import unquote from moto.core.responses import BaseResponse from .models import SESV2Backend, sesv2_backends class SESV2Response(BaseResponse): """Handler for SESV2 requests and responses.""" def __init__(self) -> None: super().__init__(service_name="sesv2") @property def sesv2_backend(self) -> SESV2Backend: """Return backend instance specific for this region.""" return sesv2_backends[self.current_account][self.region] def send_email(self) -> str: """Piggy back on functionality from v1 mostly""" params = json.loads(self.body) from_email_address = params.get("FromEmailAddress") destination = params.get("Destination", {}) content = params.get("Content") if "Raw" in content: all_destinations: List[str] = [] if "ToAddresses" in destination: all_destinations = all_destinations + destination["ToAddresses"] if "CcAddresses" in destination: all_destinations = all_destinations + destination["CcAddresses"] if "BccAddresses" in destination: all_destinations = all_destinations + destination["BccAddresses"] message = self.sesv2_backend.send_raw_email( source=from_email_address, destinations=all_destinations, raw_data=base64.b64decode(content["Raw"]["Data"]).decode("utf-8"), ) elif "Simple" in content: content_body = content["Simple"]["Body"] if "Html" in content_body: body = content_body["Html"]["Data"] else: body = content_body["Text"]["Data"] message = self.sesv2_backend.send_email( # type: ignore source=from_email_address, destinations=destination, subject=content["Simple"]["Subject"]["Data"], body=body, ) elif "Template" in content: raise NotImplementedError("Template functionality not ready") return json.dumps({"MessageId": message.id}) def create_contact_list(self) -> str: params = json.loads(self.body) self.sesv2_backend.create_contact_list(params) return json.dumps({}) def get_contact_list(self) -> str: contact_list_name = self._get_param("ContactListName") contact_list = self.sesv2_backend.get_contact_list(contact_list_name) return json.dumps(contact_list.response_object) def list_contact_lists(self) -> str: contact_lists = self.sesv2_backend.list_contact_lists() return json.dumps(dict(ContactLists=[c.response_object for c in contact_lists])) def delete_contact_list(self) -> str: name = self._get_param("ContactListName") self.sesv2_backend.delete_contact_list(name) return json.dumps({}) def create_contact(self) -> str: contact_list_name = self._get_param("ContactListName") params = json.loads(self.body) self.sesv2_backend.create_contact(contact_list_name, params) return json.dumps({}) def get_contact(self) -> str: email = unquote(self._get_param("EmailAddress")) contact_list_name = self._get_param("ContactListName") contact = self.sesv2_backend.get_contact(email, contact_list_name) return json.dumps(contact.response_object) def list_contacts(self) -> str: contact_list_name = self._get_param("ContactListName") contacts = self.sesv2_backend.list_contacts(contact_list_name) return json.dumps(dict(Contacts=[c.response_object for c in contacts])) def delete_contact(self) -> str: email = self._get_param("EmailAddress") contact_list_name = self._get_param("ContactListName") self.sesv2_backend.delete_contact(unquote(email), contact_list_name) return json.dumps({}) def create_email_identity(self) -> str: email_identity_name = self._get_param("EmailIdentity") tags = self._get_param("Tags") dkim_signing_attributes = self._get_param("DkimSigningAttributes") configuration_set_name = self._get_param("ConfigurationSetName") email_identity = self.sesv2_backend.create_email_identity( email_identity=email_identity_name, tags=tags, dkim_signing_attributes=dkim_signing_attributes, configuration_set_name=configuration_set_name, ) return json.dumps( dict( IdentityType=email_identity.identity_type, VerifiedForSendingStatus=email_identity.verified_for_sending_status, DkimAttributes=email_identity.dkim_attributes, ) ) def get_email_identity(self) -> str: email_identity_name = self._get_param("EmailIdentity") email_identity = self.sesv2_backend.get_email_identity( email_identity=email_identity_name, ) return json.dumps(email_identity.get_response_object) def list_email_identities(self) -> str: next_token = self._get_param("NextToken") page_size = self._get_param("PageSize") email_identities, next_token = self.sesv2_backend.list_email_identities( next_token=next_token, page_size=page_size, ) if isinstance(email_identities, list): response = [e.list_response_object for e in email_identities] else: response = [] return json.dumps( dict( EmailIdentities=response, NextToken=next_token, ) ) def create_configuration_set(self) -> str: configuration_set_name = self._get_param("ConfigurationSetName") tracking_options = self._get_param("TrackingOptions") delivery_options = self._get_param("DeliveryOptions") reputation_options = self._get_param("ReputationOptions") sending_options = self._get_param("SendingOptions") tags = self._get_param("Tags") suppression_options = self._get_param("SuppressionOptions") vdm_options = self._get_param("VdmOptions") self.sesv2_backend.create_configuration_set( configuration_set_name=configuration_set_name, tracking_options=tracking_options, delivery_options=delivery_options, reputation_options=reputation_options, sending_options=sending_options, tags=tags, suppression_options=suppression_options, vdm_options=vdm_options, ) return json.dumps({}) def delete_configuration_set(self) -> str: configuration_set_name = self._get_param("ConfigurationSetName") self.sesv2_backend.delete_configuration_set( configuration_set_name=configuration_set_name, ) return json.dumps({}) def get_configuration_set(self) -> str: configuration_set_name = self._get_param("ConfigurationSetName") config_set = self.sesv2_backend.get_configuration_set( configuration_set_name=configuration_set_name, ) return json.dumps(config_set.to_dict_v2()) def list_configuration_sets(self) -> str: next_token = self._get_param("NextToken") page_size = self._get_param("PageSize") configuration_sets = self.sesv2_backend.list_configuration_sets( next_token=next_token, page_size=page_size ) return json.dumps( dict(ConfigurationSets=configuration_sets, NextToken=next_token) ) def create_dedicated_ip_pool(self) -> str: pool_name = self._get_param("PoolName") tags = self._get_param("Tags") scaling_mode = self._get_param("ScalingMode") self.sesv2_backend.create_dedicated_ip_pool( pool_name=pool_name, tags=tags, scaling_mode=scaling_mode, ) return json.dumps({}) def delete_dedicated_ip_pool(self) -> str: pool_name = self._get_param("PoolName") self.sesv2_backend.delete_dedicated_ip_pool( pool_name=pool_name, ) return json.dumps({}) def list_dedicated_ip_pools(self) -> str: next_token = self._get_param("NextToken") page_size = self._get_param("PageSize") dedicated_ip_pools, next_token = self.sesv2_backend.list_dedicated_ip_pools( next_token=next_token, page_size=page_size ) return json.dumps( dict(DedicatedIpPools=dedicated_ip_pools, NextToken=next_token) ) def get_dedicated_ip_pool(self) -> str: pool_name = self._get_param("PoolName") dedicated_ip_pool = self.sesv2_backend.get_dedicated_ip_pool( pool_name=pool_name, ) return json.dumps(dict(DedicatedIpPool=dedicated_ip_pool.to_dict()))