Add template manager route

Author: MU-Software

runtime/app.py

@@ -1,21 +1,11 @@
-import json
 import logging
-import typing
 
 import chalice.app
 import chalicelib.config as config_module
 import chalicelib.logger.slack as slack_logger
 import chalicelib.route
 import chalicelib.worker
 
-if typing.TYPE_CHECKING:
-    import mypy_boto3_sqs.type_defs
-
-    SQSEventType = typing.TypedDict("SQSEventType", {"Records": list["mypy_boto3_sqs.type_defs.MessageTypeDef"]})
-else:
-    SQSEventType = dict[str, typing.Any]
-
-
 config = config_module.config
 app = chalice.app.Chalice(app_name="notico")
 app.log.setLevel(logging.INFO)
@@ -26,23 +16,5 @@
     app.log.setLevel(logging.DEBUG)
     app.log.warning("Slack logger is not configured")
 
-chalicelib.route.register_route(app)
-
-
-@app.on_sqs_message(queue=config.infra.queue_name)
-def sqs_handler(event: chalice.app.SQSEvent) -> list[dict[str, typing.Any]]:
-    parsed_event: SQSEventType = event.to_dict()
-    app.log.info(f"{parsed_event=}")
-
-    results: list[dict[str, typing.Any]] = []
-    for record in parsed_event["Records"]:
-        try:
-            worker_name = json.loads(record["body"])["worker"]
-            result = chalicelib.worker.workers[worker_name](app=app, record=record)
-            results.append(result)
-        except Exception as e:
-            app.log.error(f"Failed to handle event: {record}", exc_info=e)
-            results.append({"error": "Failed to handle event"})
-
-    app.log.info(f"{results=}")
-    return results
+chalicelib.route.register_blueprints(app)
+chalicelib.worker.register_worker(app)

runtime/chalicelib/route/__init__.py

@@ -3,28 +3,23 @@
 
 import chalice.app
 import chalicelib.util.import_util as import_util
-import chalicelib.util.type_util as type_util
 
 
-def register_route(app: chalice.app.Chalice) -> None:
-    registered_routes: dict[str, set[type_util.HttpMethodType]] = {}
+def register_blueprints(app: chalice.app.Chalice) -> None:
+    blueprints: dict[str, chalice.app.Blueprint] = {}
 
     for path in pathlib.Path(__file__).parent.glob("*.py"):
         if path.stem.startswith("__") or not (
             patterns := typing.cast(
-                type_util.RouteCollectionType,
-                getattr(
-                    import_util.load_module(path),
-                    "route_patterns",
-                    None,
-                ),
+                dict[str, chalice.app.Blueprint],
+                getattr(import_util.load_module(path), "blueprints", None),
             )
         ):
             continue
 
-        for pattern, methods, handler in patterns:
-            if pattern in registered_routes and (_mtd := registered_routes[pattern] & set(methods)):
-                raise ValueError(f"Route {pattern} with methods {_mtd} is already registered")
+        for url_prefix, blueprint in patterns.items():
+            if url_prefix in blueprints:
+                raise ValueError(f"URL Prefix {url_prefix} is already registered")
 
-            app.route(path=pattern, methods=methods)(handler)
-            registered_routes.setdefault(pattern, set()).update(methods)
+            app.register_blueprint(blueprint=blueprint, url_prefix=url_prefix)
+            blueprints[url_prefix] = blueprint

runtime/chalicelib/route/health_check.py

@@ -1,15 +1,16 @@
-import chalicelib.util.type_util as type_util
+import chalice.app
 
+health_check_api = chalice.app.Blueprint(__name__)
 
+
+@health_check_api.route("/readyz", methods=["GET"])
 def readyz() -> dict[str, str]:
     return {"status": "ok"}
 
 
+@health_check_api.route("/livez", methods=["GET"])
 def livez() -> dict[str, str]:
     return {"status": "ok"}
 
 
-route_patterns: type_util.RouteInfoType = [
-    ("/readyz", ["GET"], readyz),
-    ("/livez", ["GET"], livez),
-]
+blueprints: dict[str, chalice.app.Blueprint] = {"/health": health_check_api}

runtime/chalicelib/route/index.py

@@ -1,10 +1,11 @@
-import chalicelib.util.type_util as type_util
+import chalice.app
 
+index_api = chalice.app.Blueprint(__name__)
 
+
+@index_api.route("/", methods=["GET"])
 def index() -> dict[str, str]:
     return {"service": "notico"}
 
 
-route_patterns: type_util.RouteInfoType = [
-    ("/", ["GET"], index),
-]
+blueprints: dict[str, chalice.app.Blueprint] = {"/": index_api}

runtime/chalicelib/route/template_manager.py

@@ -0,0 +1,49 @@
+import chalice
+import chalice.app
+import chalicelib.template_manager as template_manager
+import chalicelib.util.type_util as type_util
+
+template_manager_api = chalice.app.Blueprint(__name__)
+
+
+@template_manager_api.route("/", methods=["GET"])
+def list_template_manager_services() -> dict[str, list[str]]:
+    return {"template_managers": list(template_manager.template_managers.keys())}
+
+
+@template_manager_api.route("/{service_name}", methods=["GET"])
+def list_templates(service_name: str) -> dict[str, list[dict[str, str]]]:
+    if service_name not in template_manager.template_managers:
+        raise chalice.NotFoundError(f"Service {service_name} not found")
+    return {"templates": [t.model_dump(mode="json") for t in template_manager.template_managers[service_name].list()]}
+
+
+@template_manager_api.route("/{service_name}/{code}", methods=["GET", "POST", "PUT", "DELETE"])
+def crud_template(service_name: str, code: str) -> dict[str, str]:
+    request: chalice.app.Request = template_manager_api.current_request
+    method: type_util.HttpMethodType = request.method.upper()
+    payload: dict[str, str] | None = request.json_body if method in {"POST", "PUT"} else None
+
+    if not (template_mgr := template_manager.template_managers.get(service_name, None)):
+        raise chalice.NotFoundError(f"Service {service_name} not found")
+
+    if method == "GET":
+        if template_info := template_mgr.retrieve(code):
+            return template_info.model_dump(mode="json")
+        raise chalice.NotFoundError(f"Template {code} not found")
+    elif method == "POST":
+        if not payload:
+            raise chalice.BadRequestError("Payload is required")
+        return template_mgr.create(code, payload).model_dump(mode="json")
+    elif method == "PUT":
+        if not payload:
+            raise chalice.BadRequestError("Payload is required")
+        return template_mgr.update(code, payload).model_dump(mode="json")
+    elif method == "DELETE":
+        template_mgr.delete(code)
+        return {"code": code}
+
+    raise chalice.NotFoundError(f"Method {method} is not allowed")
+
+
+blueprints: dict[str, chalice.app.Blueprint] = {"/template-manager": template_manager_api}

runtime/chalicelib/template_manager/__interface__.py

@@ -5,6 +5,7 @@
 import json
 import typing
 
+import botocore.exceptions
 import chalicelib.aws_resource as aws_resource
 import chalicelib.template_manager.__interface__ as template_mgr_interface
 import chalicelib.util.type_util as type_util
@@ -62,9 +63,12 @@ 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 retrieve(self, code: str) -> template_mgr_interface.TemplateInformation | None:
+        try:
+            template_body: str = self.resource.download(code=code).decode(encoding="utf-8")
+            return template_mgr_interface.TemplateInformation(code=code, template=template_body)
+        except botocore.exceptions.ClientError:
+            return None
 
     def create(self, code: str, template_data: type_util.TemplateType) -> template_mgr_interface.TemplateInformation:
         self.check_template_valid(template_data=template_data)

runtime/chalicelib/template_manager/toast_alimtalk.py

@@ -23,11 +23,11 @@ def list(self) -> list[template_mgr_interface.TemplateInformation]:
             for t in self.client.get_template_list().templateListResponse.templates
         ]
 
-    def retrieve(self, code: str) -> template_mgr_interface.TemplateInformation:
+    def retrieve(self, code: str) -> template_mgr_interface.TemplateInformation | None:
         query_params = toast_alimtalk_client.TemplateListQueryRequest(templateCode=code)
-        if not (t := self.client.get_template_list(query_params=query_params).templateListResponse.templates):
-            raise FileNotFoundError
-        return template_mgr_interface.TemplateInformation(code=t[0].templateCode, template=t[0].templateContent)
+        if t := self.client.get_template_list(query_params=query_params).templateListResponse.templates:
+            return template_mgr_interface.TemplateInformation(code=t[0].templateCode, template=t[0].templateContent)
+        return None
 
     def create(self, code: str, template_data: str) -> template_mgr_interface.TemplateInformation:
         raise NotImplementedError("Toast 콘솔에서 직접 템플릿을 생성해주세요.")

runtime/chalicelib/util/type_util.py

@@ -3,8 +3,6 @@
 import pydantic
 
 HttpMethodType = typing.Literal["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD", "TRACE", "CONNECT"]
-RouteInfoType = tuple[str, list[HttpMethodType], typing.Callable]
-RouteCollectionType = list[RouteInfoType]
 
 AllowedBasicValueTypes = str | int | float | bool | None
 AllowedValueTypes = AllowedBasicValueTypes | list[AllowedBasicValueTypes] | dict[str, AllowedBasicValueTypes] | None

runtime/chalicelib/worker/__init__.py

@@ -1,25 +1,50 @@
+import json
 import pathlib
 import typing
 
 import chalice.app
+import chalicelib.config as config_module
 import chalicelib.util.import_util as import_util
 
-_WorkerCollectionType = dict[str, typing.Callable[[chalice.app.SQSEvent], dict[str, typing.Any]]]
+if typing.TYPE_CHECKING:
+    import mypy_boto3_sqs.type_defs
+
+    SQSEventType = typing.TypedDict("SQSEventType", {"Records": list["mypy_boto3_sqs.type_defs.MessageTypeDef"]})
+else:
+    SQSEventType = dict[str, typing.Any]
+
+_WorkerCollectionType = dict[str, typing.Callable[[chalice.app.Chalice, chalice.app.SQSEvent], dict[str, typing.Any]]]
 workers: _WorkerCollectionType = {}
 
 for _path in pathlib.Path(__file__).parent.glob("*.py"):
     if _path.stem.startswith("__") or not (
         _patterns := typing.cast(
             _WorkerCollectionType,
-            getattr(
-                import_util.load_module(_path),
-                "worker_patterns",
-                None,
-            ),
+            getattr(import_util.load_module(_path), "worker_patterns", None),
         )
     ):
         continue
 
     if _duplicated := workers.keys() & _patterns.keys():
         raise ValueError(f"Worker {_duplicated} is already registered")
     workers.update(_patterns)
+
+
+def register_worker(app: chalice.app.Chalice) -> None:
+    @app.on_sqs_message(queue=config_module.config.infra.queue_name)
+    def sqs_handler(event: chalice.app.SQSEvent) -> list[dict[str, typing.Any]]:
+        parsed_event: SQSEventType = event.to_dict()
+        app.log.info(f"{parsed_event=}")
+
+        results: list[dict[str, typing.Any]] = []
+        for record in parsed_event["Records"]:
+            try:
+                worker_name = json.loads(record["body"])["worker"]
+                result = workers[worker_name](app, record)
+                results.append(result)
+            except Exception as e:
+                app.log.error(f"Failed to handle event: {record}", exc_info=e)
+                results.append({"error": "Failed to handle event"})
+
+        app.log.info(f"{results=}")
+        return results