Add manager interfaces and implementations

Author: MU-Software

runtime/chalicelib/send_manager/__interface__.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import typing
+
+import chalicelib.template_manager.__interface__ as template_mgr_interface
+import chalicelib.util.type_util as type_util
+import pydantic
+
+
+class BaseSendRequest(pydantic.BaseModel):
+    template_code: str
+    shared_context: type_util.ContextType
+    personalized_context: dict[str, type_util.ContextType]
+
+
+class BaseSendRawRequest(pydantic.BaseModel):
+    personalized_content: dict[str, type_util.ContextType]
+
+
+SendRequestType = typing.TypeVar("SendRequestType", bound=BaseSendRequest)
+SendRawRequestType = typing.TypeVar("SendRawRequestType", bound=BaseSendRawRequest)
+TemplateManagerType = typing.TypeVar("TemplateManagerType", bound=template_mgr_interface.TemplateManagerInterface)
+
+
+class SendManagerInterface[SendRequestType, SendRawRequestType, TemplateManagerType](typing.Protocol):
+    CAN_SEND_RAW_MESSAGE: typing.ClassVar[bool] = False
+
+    @property
+    def initialized(self) -> bool: ...
+
+    @property
+    def template_manager(self) -> TemplateManagerType: ...
+
+    def send(self, request: SendRequestType) -> dict[str, str | None]: ...
+
+    def send_raw(self, request: SendRawRequestType) -> dict[str, str | None]: ...

runtime/chalicelib/send_manager/aws_ses.py

@@ -0,0 +1,70 @@
+import traceback
+
+import botocore.exceptions
+import chalicelib.aws_resource as aws_resource_module
+import chalicelib.send_manager.__interface__ as sendmgr_interface
+import chalicelib.template_manager.aws_ses as aws_ses_template_mgr
+
+
+class AWSSESSendRawRequest(sendmgr_interface.BaseSendRawRequest):
+    personalized_content: dict[str, aws_ses_template_mgr.EmailTemplateManager.TemplateStructure]
+
+
+class AWSSESSendManager(
+    sendmgr_interface.SendManagerInterface[
+        sendmgr_interface.BaseSendRequest, AWSSESSendRawRequest, aws_ses_template_mgr.EmailTemplateManager
+    ]
+):
+    CAN_SEND_RAW_MESSAGE: bool = True
+
+    @property
+    def initialized(self) -> bool:
+        return True
+
+    @property
+    def template_manager(self) -> aws_ses_template_mgr.EmailTemplateManager:
+        return aws_ses_template_mgr.email_template_manager
+
+    def _send_email(self, from_: str, to_: str, title: str, body: str) -> str:
+        try:
+            return aws_resource_module.ses_client.send_email(
+                Source=from_,
+                # Because if you send it to multiple people at once,
+                # the e-mail addresses of the people you send with might be exposed to each other.
+                # So, we send it one by one.
+                Destination={"ToAddresses": [to_]},
+                Message={
+                    "Subject": {"Charset": "UTF-8", "Data": title},
+                    "Body": {"Html": {"Charset": "UTF-8", "Data": body}},
+                },
+            )["MessageId"]
+        except Exception as e:
+            err_tb = "\n".join(traceback.format_exception(e))
+            if isinstance(e, botocore.exceptions.HTTPClientError):
+                return e.response.get("Error", {}).get("Message", err_tb)
+            return err_tb
+
+    def send(self, request: sendmgr_interface.SendRequestType) -> dict[str, str]:
+        result: dict[str, str] = {}
+
+        for receiver, personalized_context in request.personalized_context.items():
+            context = request.shared_context | personalized_context
+            render_result = self.template_manager.render(code=request.template_code, context=context)
+            result[receiver] = self._send_email(
+                from_=render_result["from_address"],
+                to_=receiver,
+                title=render_result["title"],
+                body=render_result["body"],
+            )
+
+        return result
+
+    def send_raw(self, request: AWSSESSendRawRequest) -> dict[str, str | None]:
+        content_set = request.personalized_content.items()
+        return {r: self._send_email(from_=c.from_, to_=r, title=c.title, body=c.body) for r, c in content_set}
+
+
+aws_ses_send_manager = AWSSESSendManager()
+send_manager_patterns: dict[str, sendmgr_interface.SendManagerInterface] = {
+    "aws_ses": aws_ses_send_manager,
+}

runtime/chalicelib/send_manager/toast_alimtalk.py

@@ -0,0 +1,57 @@
+import functools
+import typing
+
+import chalicelib.config as config_module
+import chalicelib.external_api.toast_alimtalk as toast_alimtalk_client
+import chalicelib.send_manager.__interface__ as sendmgr_interface
+import chalicelib.template_manager.toast_alimtalk as toast_alimtalk_template_mgr
+
+
+class ToastAlimtalkSendRequest(sendmgr_interface.BaseSendRequest):
+    def as_request_payload(self) -> toast_alimtalk_client.MsgSendRequest:
+        return toast_alimtalk_client.MsgSendRequest(
+            senderKey=config_module.config.toast.sender_key.get_secret_value(),
+            templateCode=self.template_code,
+            recipientList=[
+                toast_alimtalk_client.MsgSendRequest.Recipient(
+                    recipientNo=send_to,
+                    templateParameter=personalized_data,
+                )
+                for send_to, personalized_data in self.personalized_context.items()
+            ],
+        )
+
+
+class ToastAlimtalkSendManager(
+    sendmgr_interface.SendManagerInterface[
+        ToastAlimtalkSendRequest,
+        sendmgr_interface.BaseSendRawRequest,
+        toast_alimtalk_template_mgr.ToastAlimtalkTemplateManager,
+    ]
+):
+    CAN_SEND_RAW_MESSAGE: bool = True
+
+    @property
+    def initialized(self) -> bool:
+        return config_module.config.toast.is_configured()
+
+    @property
+    def template_manager(self) -> toast_alimtalk_template_mgr.ToastAlimtalkTemplateManager:
+        return toast_alimtalk_template_mgr.toast_alimtalk_template_manager
+
+    @functools.cached_property
+    def client(self) -> toast_alimtalk_client.ToastAlimTalkClient:
+        return toast_alimtalk_client.ToastAlimTalkClient()
+
+    def send(self, request: ToastAlimtalkSendRequest) -> dict[str, str]:
+        request_payload = request.as_request_payload()
+        return {r.recipientNo: r.resultCode for r in self.client.send_alimtalk(request_payload).message.sendResults}
+
+    def send_raw(self, request: sendmgr_interface.BaseSendRawRequest) -> typing.NoReturn:
+        raise NotImplementedError("Toast 알림톡은 Raw 메시지를 보낼 수 없습니다.")
+
+
+toast_alimtalk_send_manager = ToastAlimtalkSendManager()
+send_manager_patterns: dict[str, sendmgr_interface.SendManagerInterface] = {
+    "toast_alimtalk": toast_alimtalk_send_manager,
+}

runtime/chalicelib/template_manager/__init__.py

@@ -0,0 +1,26 @@
+import pathlib
+import typing
+
+import chalicelib.template_manager.__interface__ as template_mgr_interface
+import chalicelib.util.import_util as import_util
+
+_TemplateManagerCollectionType = dict[str, template_mgr_interface.TemplateManagerInterface]
+template_managers: _TemplateManagerCollectionType = {}
+
+for _path in pathlib.Path(__file__).parent.glob("*.py"):
+    if _path.stem.startswith("__") or not (
+        _patterns := typing.cast(
+            _TemplateManagerCollectionType,
+            getattr(
+                import_util.load_module(_path),
+                "template_manager_patterns",
+                None,
+            ),
+        )
+    ):
+        continue
+
+    if _duplicated := template_managers.keys() & _patterns.keys():
+        raise ValueError(f"Template manager {_duplicated} is already registered")
+
+    template_managers.update(_patterns)

runtime/chalicelib/template_manager/__interface__.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import dataclasses
+import functools
+import json
+import typing
+
+import chalicelib.aws_resource as aws_resource
+import chalicelib.template_manager.__interface__ as template_mgr_interface
+import chalicelib.util.type_util as type_util
+import jinja2
+import jinja2.meta
+import jinja2.nodes
+import pydantic
+
+
+class TemplateInformation(pydantic.BaseModel):
+    code: str
+    template: type_util.TemplateType
+
+    @pydantic.computed_field  # type: ignore[misc]
+    @property
+    def template_variables(self) -> set[str]:
+        # From https://stackoverflow.com/a/77363330
+        template = jinja2.Environment(autoescape=True).parse(source=json.dumps(self.template))
+        return jinja2.meta.find_undeclared_variables(ast=template)
+
+
+TemplateStructureType = typing.TypeVar("TemplateStructureType", bound=pydantic.BaseModel)
+
+
+class TemplateManagerInterface(typing.Protocol[TemplateStructureType]):
+    template_structure_cls: type[TemplateStructureType] | type[str] | None = None
+
+    def check_template_valid(self, template_data: type_util.TemplateType) -> bool:
+        if not self.template_structure_cls or self.template_structure_cls == str:
+            return True
+        typing.cast(type[TemplateStructureType], self.template_structure_cls).model_validate(template_data)
+        return True
+
+    def list(self) -> list[TemplateInformation]: ...
+
+    def retrieve(self, code: str) -> TemplateInformation | None: ...
+
+    def create(self, code: str, template_data: type_util.TemplateType) -> TemplateInformation: ...
+
+    def update(self, code: str, template_data: type_util.TemplateType) -> TemplateInformation: ...
+
+    def delete(self, code: str) -> None: ...
+
+    def render(self, code: str, context: type_util.ContextType) -> type_util.TemplateType: ...
+
+
+S3TemplateStructureType = typing.TypeVar("S3TemplateStructureType", bound=pydantic.BaseModel)
+
+
+@dataclasses.dataclass
+class S3ResourceTemplateManager(template_mgr_interface.TemplateManagerInterface):
+    resource: typing.ClassVar[aws_resource.S3ResourcePath]
+
+    def list(self) -> list[template_mgr_interface.TemplateInformation]:
+        return [self.retrieve(code=f.split(sep=".")[0]) for f in self.resource.list_objects(filter_by_extension=True)]
+
+    @functools.lru_cache  # noqa: B019
+    def retrieve(self, code: str) -> template_mgr_interface.TemplateInformation:
+        template_body: str = self.resource.download(code=code).decode(encoding="utf-8")
+        return template_mgr_interface.TemplateInformation(code=code, template=template_body)
+
+    def create(self, code: str, template_data: type_util.TemplateType) -> template_mgr_interface.TemplateInformation:
+        self.check_template_valid(template_data=template_data)
+        self.resource.upload(code=code, content=template_data)
+        return template_mgr_interface.TemplateInformation(code=code, template=template_data)
+
+    def update(self, code: str, template_data: type_util.TemplateType) -> template_mgr_interface.TemplateInformation:
+        return self.create(code=code, template_data=template_data)
+
+    def delete(self, code: str) -> None:
+        self.resource.delete(code=code)
+
+    def render(self, code: str, context: type_util.ContextType) -> type_util.TemplateType:
+        return json.loads(jinja2.Template(source=json.dumps(obj=self.retrieve(code=code).template)).render(context))

runtime/chalicelib/template_manager/aws_ses.py

@@ -0,0 +1,19 @@
+import chalicelib.aws_resource as aws_resource
+import chalicelib.template_manager.__interface__ as template_mgr_interface
+import pydantic
+
+
+class EmailTemplateManager(template_mgr_interface.S3ResourceTemplateManager):
+    class TemplateStructure(pydantic.BaseModel):
+        from_: pydantic.EmailStr
+        title: str
+        body: str
+
+    template_structure_cls = TemplateStructure
+    resource = aws_resource.S3ResourcePath.email_template
+
+
+email_template_manager = EmailTemplateManager()
+template_manager_patterns: dict[str, template_mgr_interface.TemplateManagerInterface] = {
+    "email": email_template_manager,
+}

runtime/chalicelib/template_manager/firebase_cloudmessaging.py

@@ -0,0 +1,18 @@
+import chalicelib.aws_resource as aws_resource
+import chalicelib.template_manager.__interface__ as template_mgr_interface
+import pydantic
+
+
+class FirebaseTemplateManager(template_mgr_interface.S3ResourceTemplateManager):
+    class TemplateStructure(pydantic.BaseModel):
+        title: str
+        body: str
+
+    template_structure_cls = TemplateStructure
+    resource = aws_resource.S3ResourcePath.firebase_template
+
+
+firebase_cloudmessaging_template_manager = FirebaseTemplateManager()
+template_manager_patterns: dict[str, template_mgr_interface.TemplateManagerInterface] = {
+    "firebase_cloudmessaging": firebase_cloudmessaging_template_manager,
+}