Refactor some code

Author: MU-Software

cdk.py

@@ -198,7 +198,7 @@ def __init__(
 
 
 if __name__ == "__main__":
-    config = config_module.get_config()
+    config = config_module.config
     app = aws_cdk.App()
     notico_queue = NoticoQueue(scope=app, id="notico-queue", config=config)
     notico_ecr = NotiCoEcr(scope=app, id="notico-ecr", config=config)

runtime/app.py

@@ -3,7 +3,7 @@
 import typing
 
 import chalice.app
-import chalicelib.config
+import chalicelib.config as config_module
 import chalicelib.logger.slack as slack_logger
 import chalicelib.route
 import chalicelib.worker
@@ -16,7 +16,7 @@
     SQSEventType = dict[str, typing.Any]
 
 
-config = chalicelib.config.get_config()
+config = config_module.config
 app = chalice.app.Chalice(app_name="notico")
 app.log.setLevel(logging.INFO)
 
@@ -38,7 +38,7 @@ def sqs_handler(event: chalice.app.SQSEvent) -> 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, config=config, record=record)
+            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)

runtime/chalicelib/aws_resource.py

@@ -1,8 +1,10 @@
+import dataclasses
 import enum
+import functools
 import typing
 
 import boto3
-import chalicelib.config
+import chalicelib.config as config_module
 
 if typing.TYPE_CHECKING:
     import mypy_boto3_s3.client
@@ -15,11 +17,37 @@
 s3_client: "mypy_boto3_s3.client.S3Client" = boto3.client(service_name="s3")
 
 
-class S3ResourcePath(enum.StrEnum):
-    email_template = "email/template/{filename}.html"
+@dataclasses.dataclass
+class S3ResourceInfo:
+    prefix: str
+    extension: str
 
-    def as_path(self, filename: str) -> str:
-        return self.value.format(filename=filename)
+    def as_path(self, code: str) -> str:
+        return self.prefix + f"{code}.{self.extension}"
 
-    def download(self, conf: chalicelib.config.Config, filename: str) -> bytes:
-        return s3_client.get_object(Bucket=conf.infra.s3_bucket_name, Key=self.as_path(filename))["Body"].read()
+
+class S3ResourcePath(enum.Enum):
+    email_template = S3ResourceInfo(prefix="email/template/", extension="json")
+    telegram_template = S3ResourceInfo(prefix="telegram/template/", extension="json")
+    firebase_template = S3ResourceInfo(prefix="firebase/template/", extension="json")
+
+    @functools.cached_property
+    def s3_bucket_name(self) -> config_module.InfraConfig:
+        return config_module.config.infra.s3_bucket_name
+
+    def download(self, code: str) -> bytes:
+        return s3_client.get_object(Bucket=self.s3_bucket_name, Key=self.value.as_path(code))["Body"].read()
+
+    def upload(self, code: str, content: str) -> None:
+        s3_client.put_object(Bucket=self.s3_bucket_name, Key=self.value.as_path(code), Body=content)
+
+    def delete(self, code: str) -> None:
+        s3_client.delete_object(Bucket=self.s3_bucket_name, Key=self.value.as_path(code))
+
+    def list_objects(self, filter_by_extension: bool = False) -> list[str]:
+        return [
+            key
+            for obj in s3_client.list_objects(Bucket=self.s3_bucket_name, Prefix=self.value.prefix)["Contents"]
+            if (key := obj["Key"].replace(self.value.prefix, ""))
+            and (not filter_by_extension or key.split(".")[-1] == self.value.extension)
+        ]

runtime/chalicelib/config.py

@@ -1,4 +1,4 @@
-import functools
+import logging
 import typing
 import urllib.parse
 
@@ -7,6 +7,16 @@
 import pydantic_settings
 
 AllowedToastServices = typing.Literal["alimtalk"]
+logger = logging.getLogger(__name__)
+
+
+def log_request(req: httpx.Request) -> None:
+    logger.debug(f"REQ [{req.method}]{req.url}")
+
+
+def log_response(resp: httpx.Response) -> None:
+    req = resp.request
+    logger.debug(f"RES [{req.method}]{req.url}<{resp.status_code}> {resp.content.decode(errors='ignore')=}")
 
 
 class InfraConfig(pydantic_settings.BaseSettings):
@@ -46,6 +56,7 @@ def get_session(self, service: AllowedToastServices) -> httpx.Client:
                 "Content-Type": "application/json;charset=UTF-8",
             },
             timeout=self.timeout,
+            event_hooks={"request": [log_request], "response": [log_response]},
         )
 
 
@@ -55,6 +66,15 @@ class FirebaseConfig(_ServiceConfig, pydantic_settings.BaseSettings):
 
 class TelegramConfig(_ServiceConfig, pydantic_settings.BaseSettings):
     bot_token: pydantic.SecretStr | None = None
+    timeout: float = 3.0
+
+    def get_session(self) -> httpx.Client:
+        return httpx.Client(
+            base_url=f"https://api.telegram.org/bot{self.bot_token.get_secret_value()}/",
+            headers={"Content-Type": "application/json;charset=UTF-8"},
+            timeout=self.timeout,
+            event_hooks={"request": [log_request], "response": [log_response]},
+        )
 
 
 class SlackConfig(_ServiceConfig, pydantic_settings.BaseSettings):
@@ -72,6 +92,4 @@ class Config(pydantic_settings.BaseSettings):
     env_vars: dict[str, str] = pydantic.Field(default_factory=dict)
 
 
-@functools.lru_cache(maxsize=1)
-def get_config() -> Config:
-    return Config(_env_nested_delimiter="__", _case_sensitive=False)
+config = Config(_env_nested_delimiter="__", _case_sensitive=False)

runtime/chalicelib/sender/__interface__.py

@@ -1,4 +1,3 @@
-import chalicelib.config as config_module
 import pydantic
 
 ReceiverType = str
@@ -15,5 +14,3 @@ class NotificationSendRequest(pydantic.BaseModel):
     template_code: str
     shared_context: ContextType
     personalized_context: dict[ReceiverType, ContextType]
-
-    conf: config_module.Config

runtime/chalicelib/sender/aws_ses.py

@@ -61,8 +61,7 @@ def validate_shared_context(cls, v: sender_interface.ContextType) -> sender_inte
     def to_aws_ses_request(self) -> list[AWSSESRequest]:
         title_template = self.shared_context["title"]
         body_template = chalicelib.aws_resource.S3ResourcePath.email_template.download(
-            conf=self.conf,
-            filename=self.template_code,
+            code=self.template_code,
         ).decode("utf-8")
 
         return [

runtime/chalicelib/sender/firebase_cloudmessaging.py

@@ -4,7 +4,7 @@
 import logging
 import typing
 
-import chalicelib.config
+import chalicelib.config as config_module
 import firebase_admin
 import firebase_admin.credentials
 import firebase_admin.messaging
@@ -51,8 +51,6 @@ class FirebaseCloudMessaging(pydantic.BaseModel):
     webpush: firebase_admin.messaging.WebpushConfig | None = None
     fcm_options: firebase_admin.messaging.FCMOptions | None = None
 
-    conf: chalicelib.config.FirebaseConfig
-
     model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
 
     @pydantic.field_validator("data", mode="before")
@@ -88,13 +86,13 @@ def message_payloads(self) -> list[firebase_admin.messaging.Message]:
         ]
 
     def send(self) -> str:
-        if not self.conf.is_configured():
+        if not config_module.config.firebase.is_configured():
             raise ValueError("Firebase configuration is not set up properly.")
 
         with contextlib.suppress(ValueError):
             firebase_admin.initialize_app(
                 credential=firebase_admin.credentials.Certificate(
-                    cert=self.conf.certificate.get_secret_value(),
+                    cert=config_module.config.firebase.certificate.get_secret_value(),
                 )
             )