added template email sending and template init on ses

Author: Mayank808

backend/app/interfaces/email_service.py

@@ -1,4 +1,38 @@
+import json
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class TemplateData(ABC):
+    def get_formatted_string(self) -> str:
+        class_dict = self.__dict__
+        try:
+            formatted_string = json.dumps(class_dict)  # Try to convert to a JSON string
+        except (TypeError, ValueError) as e:
+            # Handle errors and return a message instead
+            return f"Error in converting data to JSON: {e}"
+
+        return formatted_string
+
+
+@dataclass
+class TestEmailData(TemplateData):
+    name: str
+    date: str
+
+
+class EmailTemplate(Enum):
+    TEST = "Test"
+
+
+@dataclass
+class EmailContent(Generic[T]):
+    recipient: str
+    data: T
 
 
 class IEmailService(ABC):
@@ -8,7 +42,7 @@ class IEmailService(ABC):
     """
 
     @abstractmethod
-    def send_email(self, subject: str, recipient: str, body_html: str) -> None:
+    def send_email(self, template: EmailTemplate, content: EmailContent) -> dict:
         """
         Sends an email with the given parameters.
 

backend/app/interfaces/email_service_provider.py

@@ -1,5 +1,7 @@
 from abc import ABC, abstractmethod
 
+from app.interfaces.email_service import EmailContent, EmailTemplate
+
 
 class IEmailServiceProvider(ABC):
     """
@@ -8,7 +10,7 @@ class IEmailServiceProvider(ABC):
     """
 
     @abstractmethod
-    def send_email(self, recipient: str, subject: str) -> None:
+    def send_email(self, template: EmailTemplate, content: EmailContent) -> dict:
         """
         Sends an email using the provider's service.
 

backend/app/routes/email.py

@@ -1,9 +1,13 @@
-import logging
 from typing import Annotated
 
 from fastapi import APIRouter, Depends
 
-from app.interfaces.email_service import IEmailService
+from app.interfaces.email_service import (
+    EmailContent,
+    EmailTemplate,
+    IEmailService,
+    TestEmailData,
+)
 from app.services.email.email_service import EmailService
 from app.services.email.email_service_provider import (
     get_email_service_provider,
@@ -14,11 +18,9 @@
     tags=["email"],
 )
 
-log = logging.getLogger("uvicorn")
-
 
 def get_email_service() -> IEmailService:
-    return EmailService(get_email_service_provider())
+    return EmailService(provider=get_email_service_provider())
 
 
 # TODO (Mayank, Nov 30th) - Remove test emails once email service is fully implemented
@@ -28,9 +30,9 @@ async def send_welcome_email(
     user_name: str,
     email_service: Annotated[IEmailService, Depends(get_email_service)],
 ):
-    log.info(f"Main Sending welcome email to {user_name} at {recipient}")
-    email_service.send_email(
-        subject="Welcome to the app!",
-        recipient=recipient,
+    return email_service.send_email(
+        template=EmailTemplate.TEST,
+        content=EmailContent[TestEmailData](
+            recipient=recipient, data=TestEmailData(name=user_name, date="2021-12-01")
+        ),
     )
-    return {"message": f"Welcome email sent to {user_name}!"}

backend/app/server.py

@@ -10,14 +10,17 @@
 load_dotenv()
 
 # we need to load env variables before initialization code runs
+# from . import models  # noqa: E402
 from .routes import user  # noqa: E402
+from .utilities.ses.ses_init import ensure_ses_templates  # noqa: E402
 
 log = logging.getLogger("uvicorn")
 
 
 @asynccontextmanager
 async def lifespan(_: FastAPI):
     log.info("Starting up...")
+    ensure_ses_templates()
     # models.run_migrations()
     # initialize_firebase()
     yield

backend/app/services/email/email_service.py

@@ -1,6 +1,6 @@
 import logging
 
-from app.interfaces.email_service import IEmailService
+from app.interfaces.email_service import EmailContent, EmailTemplate, IEmailService, T
 from app.interfaces.email_service_provider import IEmailServiceProvider
 
 
@@ -10,6 +10,25 @@ def __init__(self, provider: IEmailServiceProvider):
         self.provider = provider
         self.logger = logging.getLogger(__name__)
 
-    def send_email(self, subject: str, recipient: str, body_html: str = "") -> None:
-        self.logger.info(f"Sending email to {recipient} with subject: {subject}")
-        self.provider.send_email(subject, recipient)
+    # def render_templates(self, template: EmailTemplate) -> tuple[str, str]:
+    #     html_path = self.template_dir / f"{template.value}.html"
+    #     text_path = self.template_dir / f"{template.value}.txt"
+
+    #      # Check if both files exist
+    #     if not html_path.exists():
+    #         raise FileNotFoundError(f"HTML template not found: {html_path}")
+    #     if not text_path.exists():
+    #         raise FileNotFoundError(f"Text template not found: {text_path}")
+
+    #     # Read the templates
+    #     html_template = html_path.read_text(encoding="utf-8")
+    #     text_template = text_path.read_text(encoding="utf-8")
+
+    #     return html_template, text_template
+
+    def send_email(self, template: EmailTemplate, content: EmailContent[T]) -> dict:
+        self.logger.info(
+            f"Sending email to {content.recipient} with template {template.value}"
+        )
+        # html_template, text_template = self.render_templates(template)
+        return self.provider.send_email(template, content)

backend/app/services/email/email_service_provider.py

@@ -1,9 +1,9 @@
 import os
 
 import boto3
-from botocore.exceptions import BotoCoreError, ClientError
-from fastapi import HTTPException
+from botocore.exceptions import ClientError
 
+from app.interfaces.email_service import EmailContent, EmailTemplate
 from app.interfaces.email_service_provider import IEmailServiceProvider
 
 
@@ -25,33 +25,42 @@ def __init__(
             aws_secret_access_key=aws_secret_key,
         )
 
+        self.verified_emails = None
+        if self.is_sandbox:
+            response = self.ses_client.list_verified_email_addresses()
+            self.verified_emails = response.get("VerifiedEmailAddresses", [])
+
     def _verify_email(self, email: str):
         if not self.is_sandbox:
             return
         try:
-            self.client.verify_email_identity(EmailAddress=email)
-            print(f"Verification email sent to {email}.")
+            if email not in self.verified_emails:
+                self.ses_client.verify_email_identity(EmailAddress=email)
+                print(f"Verification email sent to {email}.")
         except Exception as e:
             print(f"Failed to verify email: {e}")
 
-    def send_email(self, subject: str, recipient: str) -> None:
+    def send_email(self, template: EmailTemplate, content: EmailContent) -> dict:
         try:
-            self._verify_email(recipient)
-            self.ses_client.send_email(
+            self._verify_email(content.recipient)
+
+            self.ses_client.get_template(TemplateName=template.value)
+
+            template_data = content.data.get_formatted_string()
+
+            response = self.ses_client.send_templated_email(
                 Source=self.source_email,
-                Destination={"ToAddresses": [recipient]},
-                Message={
-                    "Subject": {"Data": subject},
-                    "Body": {"Text": {"Data": "Hello, this is a test email!"}},
-                },
+                Destination={"ToAddresses": [content.recipient]},
+                Template=template.value,
+                TemplateData=template_data,
             )
-        except BotoCoreError as e:
-            raise HTTPException(status_code=500, detail=f"SES BotoCoreError: {e}")
+
+            return {
+                "message": "Email sent successfully!",
+                "message_id": response["MessageId"],
+            }
         except ClientError as e:
-            raise HTTPException(
-                status_code=500,
-                detail=f"SES ClientError: {e.response['Error']['Message']}",
-            )
+            return {"error": f"An error occurred: {e.response['Error']['Message']}"}
 
 
 def get_email_service_provider() -> IEmailServiceProvider:

backend/app/services/email/email_templates/test.html

@@ -0,0 +1,82 @@
+import json
+import os
+from typing import Dict
+
+import boto3
+from botocore.exceptions import ClientError
+
+TEMPLATES_FILE = "app/utilities/ses/ses_templates.json"
+TEMPLATES_DIR = "app/utilities/ses/template_files"
+
+ses_client = boto3.client(
+    "ses",
+    region_name=os.getenv("AWS_REGION"),
+    aws_access_key_id=os.getenv("AWS_ACCESS_KEY"),
+    aws_secret_access_key=os.getenv("AWS_SECRET_KEY"),
+)
+
+
+def load_templates_metadata(file_path: str) -> Dict:
+    try:
+        with open(file_path, "r") as file:
+            return json.load(file)
+    except FileNotFoundError:
+        print(f"Error: {file_path} not found.")
+        return []
+    except json.JSONDecodeError as e:
+        print(f"Error parsing {file_path}: {e}")
+        return []
+
+
+def load_file_content(file_path: str) -> str:
+    """Reads the content of a file."""
+    try:
+        with open(file_path, "r") as file:
+            return file.read()
+    except FileNotFoundError:
+        print(f"Error: File '{file_path}' not found.")
+        return ""
+
+
+templates_metadata = load_templates_metadata(TEMPLATES_FILE)
+
+
+# Function to create SES template
+def create_ses_template(template_metadata):
+    name = template_metadata["TemplateName"]
+    try:
+        text_part = load_file_content(template_metadata["TextPart"])
+        html_part = load_file_content(template_metadata["HtmlPart"])
+        if not text_part or not html_part:
+            print(f"Skipping template '{name}' missing content.")
+            return
+
+        template = {
+            "TemplateName": template_metadata["TemplateName"],
+            "SubjectPart": template_metadata["SubjectPart"],
+            "TextPart": text_part,
+            "HtmlPart": html_part,
+        }
+        ses_client.create_template(Template=template)
+        print(f"SES template '{name}' created successfully!")
+    except ClientError as e:
+        if e.response["Error"]["Code"] == "TemplateAlreadyExists":
+            print(f"SES template '{name}' already exists.")
+        else:
+            print(f"An error occurred while creating the SES template: {e}")
+
+
+# Ensure SES templates are available at app startup
+def ensure_ses_templates():
+    for template_metadata in templates_metadata:
+        name = template_metadata["TemplateName"]
+        try:
+            # Check if the template exists
+            ses_client.get_template(TemplateName=name)
+            print(f"SES template '{name}' already exists.")
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "TemplateDoesNotExist":
+                print(f"SES template '{name}' does not exist. Creating template...")
+                create_ses_template(template_metadata)
+            else:
+                print(f"An error occurred while checking the SES template: {e}")

backend/app/utilities/ses/ses_init.py

@@ -0,0 +1,8 @@
+[
+  {
+    "HtmlPart": "app/utilities/ses/template_files/test.html",
+    "SubjectPart": "Testing Email SES Template",
+    "TemplateName": "Test",
+    "TextPart": "app/utilities/ses/template_files/test.txt"
+  }
+]

backend/app/utilities/ses/ses_templates.json

@@ -0,0 +1,6 @@
+<html>
+<body>
+    <h1>Welcome, {{name}}!</h1>
+    <p>We are glad to have you with us. Thank you for joining us on {{date}}.</p>
+</body>
+</html>