update event

enrollments-events/app/events/emails/__init__.py

@@ -0,0 +1,102 @@
+from typing import TYPE_CHECKING, TypedDict
+
+from aws_lambda_powertools import Logger
+from layercake.dateutils import now
+from layercake.dynamodb import DynamoDBPersistenceLayer
+from layercake.email_ import Message
+from layercake.strutils import first_word, truncate_str
+
+from . import (
+    reminder_access_period_before_30_days,
+    reminder_cert_expiration_before_30_days,
+    reminder_no_access_after_3_days,
+    reminder_no_activity_after_7_days,
+)
+
+if TYPE_CHECKING:
+    from mypy_boto3_sesv2 import SESV2Client
+else:
+    SESV2Client = object
+
+Event = TypedDict('Event', {'id': str, 'sk': str})
+
+TEMPLATES = {
+    'reminder_access_period_before_30_days': {
+        'subject': reminder_access_period_before_30_days.SUBJECT,
+        'message': reminder_access_period_before_30_days.MESSAGE,
+    },
+    'reminder_cert_expiration_before_30_days': {
+        'subject': reminder_cert_expiration_before_30_days.SUBJECT,
+        'message': reminder_cert_expiration_before_30_days.MESSAGE,
+    },
+    'reminder_no_access_after_3_days': {
+        'subject': reminder_no_access_after_3_days.SUBJECT,
+        'message': reminder_no_access_after_3_days.MESSAGE,
+    },
+    'reminder_no_activity_after_7_days': {
+        'subject': reminder_no_activity_after_7_days.SUBJECT,
+        'message': reminder_no_activity_after_7_days.MESSAGE,
+    },
+}
+
+logger = Logger(__name__)
+
+
+def send_email(
+    to: tuple[str, str],
+    subject: str,
+    message: str,
+    context: dict = {},
+    *,
+    sender: tuple[str, str],
+    event: Event,
+    sesv2_client: SESV2Client,
+    dynamodb_persistence_layer: DynamoDBPersistenceLayer,
+) -> bool:
+    now_ = now()
+    name, _ = to
+    event_name = event['sk']
+    emailmsg = Message(
+        from_=sender,
+        to=to,
+        subject=subject.format(
+            course=truncate_str(context['course']),
+        ),
+    )
+    emailmsg.add_alternative(
+        message.format(
+            first_name=first_word(name),
+            course=context['course'],
+        )
+    )
+
+    try:
+        sesv2_client.send_email(
+            Content={
+                'Raw': {
+                    'Data': emailmsg.as_bytes(),
+                },
+            }
+        )
+        dynamodb_persistence_layer.put_item(
+            item={
+                'id': event['id'],
+                'sk': f'{event_name}#EXECUTED',
+                'created_at': now_,
+            }
+        )
+        logger.info('Email sent')
+    except Exception as exc:
+        logger.exception(exc)
+
+        dynamodb_persistence_layer.put_item(
+            item={
+                'id': event['id'],
+                'sk': f'{event_name}#FAILED',
+                'created_at': now_,
+            }
+        )
+
+        return False
+    else:
+        return True