Enhanced email functions (#498)

enhanced email sending function
---------

Signed-off-by: Trey <[email protected]>

backend/api/emails/send.py

@@ -21,14 +21,9 @@
 from backend.models import QuotaLimit
 from backend.models import QuotaUsage
 from backend.types.emails import (
-    SingleEmailInput,
     BulkEmailEmailItem,
-    BulkEmailSuccessResponse,
-    BulkEmailErrorResponse,
-    BulkTemplatedEmailInput,
 )
 
-# from backend.utils.quota_limit_ops import quota_usage_check_under
 from settings.helpers import send_email, send_templated_bulk_email
 from backend.types.htmx import HtmxHttpRequest
 
@@ -100,7 +95,7 @@ def _send_bulk_email_view(request: HtmxHttpRequest) -> HttpResponse:
             )
         )
 
-    EMAIL_DATA = BulkTemplatedEmailInput(
+    EMAIL_SENT = send_templated_bulk_email(
         email_list=email_list,
         template_name="user_send_client_email",
         default_template_data={
@@ -110,14 +105,14 @@ def _send_bulk_email_view(request: HtmxHttpRequest) -> HttpResponse:
         },
     )
 
-    EMAIL_SENT: BulkEmailSuccessResponse | BulkEmailErrorResponse = send_templated_bulk_email(data=EMAIL_DATA)
-
-    if isinstance(EMAIL_SENT, BulkEmailErrorResponse):
-        messages.error(request, EMAIL_SENT.message)
+    if EMAIL_SENT.failed:
+        messages.error(request, EMAIL_SENT.error)
         return render(request, "base/toast.html")
 
+    # todo - fix
+
     EMAIL_RESPONSES: Iterator[tuple[BulkEmailEmailItem, BulkEmailEntryResultTypeDef]] = zip(
-        EMAIL_DATA.email_list, EMAIL_SENT.response.get("BulkEmailEntryResults")  # type: ignore[arg-type]
+        email_list, EMAIL_SENT.response.get("BulkEmailEntryResults")  # type: ignore[arg-type]
     )
 
     if request.user.logged_in_as_team:
@@ -183,7 +178,7 @@ def _send_single_email_view(request: HtmxHttpRequest) -> HttpResponse:
 
     message_single_line_html = message.replace("\r\n", "<br>").replace("\n", "<br>")
 
-    EMAIL_DATA = SingleEmailInput(
+    EMAIL_SENT = send_email(
         destination=email,
         subject=subject,
         content={
@@ -198,8 +193,6 @@ def _send_single_email_view(request: HtmxHttpRequest) -> HttpResponse:
         },
     )
 
-    EMAIL_SENT = send_email(data=EMAIL_DATA)
-
     aws_message_id = None
     if EMAIL_SENT.response is not None:
         aws_message_id = EMAIL_SENT.response.get("MessageId")
@@ -211,7 +204,7 @@ def _send_single_email_view(request: HtmxHttpRequest) -> HttpResponse:
         status_object.status = "pending"
     else:
         status_object.status = "failed_to_send"
-        messages.error(request, f"Failed to send the email. Error: {EMAIL_SENT.message}")
+        messages.error(request, f"Failed to send the email. Error: {EMAIL_SENT.error}")
 
     if request.user.logged_in_as_team:
         status_object.organization = request.user.logged_in_as_team

backend/api/teams/invites.py

@@ -95,11 +95,10 @@ def return_error_notif(request: HtmxHttpRequest, message: str, autohide=None):
     )
 
     send_email(
-        SingleEmailInput(
-            destination=user.email,
-            subject="New Organization Invite",
-            content=dedent(
-                f"""
+        destination=user.email,
+        subject="New Organization Invite",
+        content=dedent(
+            f"""
                 Hi {user.first_name or "User"},
 
                 {request.user.first_name or f"User {request.user.email}"} has invited you to join the organization \"{team.name}\" (#{team.id})
@@ -111,8 +110,7 @@ def return_error_notif(request: HtmxHttpRequest, message: str, autohide=None):
                 Didn't give permission to be added to this organization? You can safely ignore the email, no actions can be done on
                 behalf of you without your action.
             """
-            ),
-        )
+        ),
     )
 
     messages.success(request, "Invitation successfully sent")

backend/service/invoices/common/emails/on_create.py

@@ -0,0 +1,64 @@
+from string import Template
+from textwrap import dedent
+
+from backend.models import Invoice, InvoiceRecurringProfile, User, EmailSendStatus
+from backend.utils.dataclasses import BaseServiceResponse
+from backend.utils.service_retry import retry_handler
+from settings.helpers import send_email
+
+
+class OnCreateInvoiceEmailServiceResponse(BaseServiceResponse[str]): ...
+
+
+def on_create_invoice_service(users_email: str, invoice: Invoice) -> OnCreateInvoiceEmailServiceResponse:
+    if not users_email:
+        return OnCreateInvoiceEmailServiceResponse(error_message="User email not found")
+
+    if not invoice:
+        return OnCreateInvoiceEmailServiceResponse(error_message="Invoice not found")
+
+    email_message = dedent(
+        """
+        Hi $first_name,
+
+        The invoice #$invoice_id has been created for you to pay, due on the $due_date. Please pay at your earliest convenience.
+
+        Balance Due: $currency_symbol$amount_due $currency
+
+        Many thanks,
+        $company_name
+    """
+    )
+
+    user_data = {
+        "first_name": invoice.client_to.name.split(" ")[0] if invoice.client_to else invoice.client_name,
+        "invoice_id": invoice.id,
+        "invoice_ref": invoice.reference or invoice.invoice_number or invoice.id,
+        "due_date": invoice.date_due.strftime("%a %m %Y"),
+        "amount_due": invoice.get_total_price(),
+        "currency": invoice.currency,
+        "currency_symbol": invoice.get_currency_symbol(),
+        "product_list": [],  # todo
+        "company_name": invoice.self_company or invoice.self_name,
+    }
+
+    output: str = Template(email_message).substitute(user_data)
+
+    email_svc_response = retry_handler(
+        send_email,
+        destination=invoice.client_to.email or invoice.client_email if invoice.client_to else invoice.client_email,
+        subject=f"Invoice #{invoice.id} from {invoice.self_company or invoice.self_name}",
+        content=output,
+    )
+
+    if email_svc_response.failed:
+        return OnCreateInvoiceEmailServiceResponse(False, error_message="Failed to send email")
+
+    EmailSendStatus.objects.create(
+        status="send",
+        owner=invoice.owner,
+        recipient=users_email,
+        aws_message_id=email_svc_response.response.get("MessageId"),
+    )
+
+    return OnCreateInvoiceEmailServiceResponse(True, response="Email sent successfully")

backend/service/invoices/recurring/generation/next_invoice.py

@@ -16,6 +16,9 @@ def generate_next_invoice_service(
     issue_date: date = date.today(),
     account_defaults: DefaultValues | None = None,
 ) -> GenerateNextInvoiceServiceResponse:
+    """
+    This will generate the next single invoice based on the invoice recurring profile
+    """
 
     if not invoice_recurring_profile:
         return GenerateNextInvoiceServiceResponse(error_message="Invoice recurring profile not found")

backend/service/teams/create_user.py

@@ -34,11 +34,10 @@ def create_user_service(
     user.save()
 
     send_email(
-        SingleEmailInput(
-            destination=email,
-            subject="MyFinances | You have been invited to join an organization",
-            content=dedent(
-                f"""
+        destination=email,
+        subject="MyFinances | You have been invited to join an organization",
+        content=dedent(
+            f"""
                 Hi {user.first_name or "User"},
 
                 You have been invited by {request.user.email} to join the organization {team.name}.
@@ -57,8 +56,7 @@ def create_user_service(
                 Didn't give permission to be added to this organization? You can safely ignore the email, no actions can be done on
                 behalf of you without your permission.
             """
-            ),
-        )
+        ),
     )
 
     team.members.add(user)

backend/signals/signals.py

@@ -97,8 +97,6 @@ def send_welcome_email(sender, instance: User, created, **kwargs):
                 Verify Link: {magic_link_url}
             """
 
-            email_input = SingleEmailInput(destination=instance.email, subject="Welcome to MyFinances", content=email_message)
-
-            email = send_email(email_input)
+            email = send_email(destination=instance.email, subject="Welcome to MyFinances", content=email_message)
 
         #     User.send_welcome_email(instance)

backend/types/emails.py

@@ -1,8 +1,16 @@
 from dataclasses import dataclass
-from typing import Literal, TypedDict
+from typing import TypedDict
 
 from mypy_boto3_sesv2.type_defs import SendEmailResponseTypeDef, SendBulkEmailResponseTypeDef, BulkEmailEntryResultTypeDef
 
+from backend.utils.dataclasses import BaseServiceResponse
+
+
+class SingleEmailSendServiceResponse(BaseServiceResponse[SendEmailResponseTypeDef]): ...
+
+
+class BulkEmailSendServiceResponse(BaseServiceResponse[SendBulkEmailResponseTypeDef]): ...
+
 
 class SingleTemplatedEmailContent(TypedDict):
     template_name: str
@@ -19,19 +27,6 @@ class SingleEmailInput:
     from_address_name_prefix: str | None = None
 
 
-@dataclass(frozen=True)
-class SingleEmailSuccessResponse:
-    response: SendEmailResponseTypeDef
-    success: Literal[True] = True
-
-
-@dataclass(frozen=True)
-class SingleEmailErrorResponse:
-    message: str
-    response: SendEmailResponseTypeDef | None
-    success: Literal[False] = False
-
-
 @dataclass
 class BulkEmailEmailItem:
     destination: str
@@ -46,16 +41,3 @@ class BulkTemplatedEmailInput:
     ConfigurationSetName: str | None = None
     from_address: str | None = None
     from_address_name_prefix: str | None = None
-
-
-@dataclass(frozen=True)
-class BulkEmailSuccessResponse:
-    response: SendBulkEmailResponseTypeDef
-    success: Literal[True] = True
-
-
-@dataclass(frozen=True)
-class BulkEmailErrorResponse:
-    message: str
-    response: SendBulkEmailResponseTypeDef | None
-    success: Literal[False] = False

backend/utils/service_retry.py

@@ -0,0 +1,19 @@
+from typing import Callable, TypeVar, Generic
+from backend.utils.dataclasses import BaseServiceResponse
+
+T = TypeVar("T", bound=BaseServiceResponse)
+
+
+def retry_handler(function: Callable[..., T], *args, retry_max_attempts: int = 3, **kwargs) -> T:
+    attempts: int = 0
+
+    while attempts < retry_max_attempts:
+        response: T = function(*args, **kwargs)
+
+        if response.failed:
+            attempts += 1
+            if attempts == retry_max_attempts:
+                return response
+            continue
+        return response
+    return response

backend/views/core/auth/login.py

@@ -138,7 +138,7 @@ def send_message(
     def send_magic_link_email(self, request: HttpRequest, user: User, uuid: str, plain_token: str) -> None:
         magic_link_url = request.build_absolute_uri(reverse("auth:login magic_link verify", kwargs={"uuid": uuid, "token": plain_token}))
 
-        email: SingleEmailInput = SingleEmailInput(
+        send_email(
             destination=user.email,
             subject="Login Request",
             content=dedent(
@@ -152,7 +152,6 @@ def send_magic_link_email(self, request: HttpRequest, user: User, uuid: str, pla
         """
             ),
         )
-        send_email(email)
 
 
 class MagicLinkWaitingView(View):

backend/views/core/auth/verify.py

@@ -75,11 +75,10 @@ def resend_verification_code(request):
     magic_link_url = settings.SITE_URL + reverse("auth:login create_account verify", kwargs={"uuid": magic_link.uuid, "token": token_plain})
 
     send_email(
-        SingleEmailInput(
-            destination=email,
-            subject="Verify your email",
-            content=dedent(
-                f"""
+        destination=email,
+        subject="Verify your email",
+        content=dedent(
+            f"""
                 Hi {user.first_name if user.first_name else "User"},
 
                 Verification for your email has been requested to link this email to your MyFinances account.
@@ -88,8 +87,7 @@ def resend_verification_code(request):
                 If it was you, you can complete the verification by clicking the link below.
                 Verify Link: {magic_link_url}
             """
-            ),
-        )
+        ),
     )
 
     messages.success(request, "Verification email sent, check your inbox or spam!")