Remove .chalice/config.json and generate it dynamically

Author: MU-Software

Makefile

@@ -52,9 +52,14 @@ lint: hooks-lint  # alias
 
 mypy: hooks-mypy  # alias
 
+# =============================================================================
+# Config related commands
+generate-chalice-config:
+	@cd $(PROJECT_DIR)/runtime && poetry run python -m chalicelib.cli generate-chalice-config
+
 # =============================================================================
 # AWS CDK related commands
-stack-ecr-deploy:
+stack-ecr-deploy: generate-chalice-config
 	@cdk deploy notico-ecr
 
 stack-queue-deploy:
@@ -69,7 +74,7 @@ stack-lambda-deploy:
 cleanup-deploy:
 	@rm -rf $(PROJECT_DIR)/cdk.out
 	@rm -rf $(PROJECT_DIR)/chalice.out
-	@rm -rf $(PROJECT_DIR)/runtime/.chalice/deployments
+	@rm -rf $(PROJECT_DIR)/runtime/.chalice
 
 stack-deploy: docker-build-prod stack-queue-deploy stack-s3-deploy stack-lambda-deploy cleanup-deploy
 

cdk.py

@@ -182,7 +182,7 @@ def __init__(
             ecr_repo=ecr_repo,
             stage_config={
                 "automatic_layer": True,
-                "environment_variables": config.env_vars,
+                "environment_variables": config.env_dict,
             },
         )
         app_default_role = app.get_role("DefaultRole")

runtime/.chalice/config.json

@@ -0,0 +1,16 @@
+import pathlib
+import typing
+
+import chalicelib.util.import_util as import_util
+import typer
+
+typer_app = typer.Typer()
+
+_cli_patterns: list[list[typing.Callable]] = import_util.auto_import_patterns(
+    pattern="cli_funcs",
+    file_prefix="",
+    dir=pathlib.Path(__file__).parent,
+)
+for _cli_funcs in _cli_patterns:
+    for _cli_func in _cli_funcs:
+        typer_app.command()(_cli_func)

runtime/chalicelib/cli/__init__.py

@@ -0,0 +1,4 @@
+import chalicelib.cli
+
+if __name__ == "__main__":
+    chalicelib.cli.typer_app()

runtime/chalicelib/cli/__main__.py

@@ -0,0 +1,5 @@
+def initial() -> None:
+    pass
+
+
+cli_funcs = [initial]

runtime/chalicelib/cli/dummy.py

@@ -0,0 +1,43 @@
+import hashlib
+import pathlib
+
+import chalicelib.config as config_module
+
+BUF_SIZE = 65536
+
+CLI_PATH = pathlib.Path(__file__).parent
+RUNTIME_PATH = CLI_PATH.parent.parent
+CHALICE_CONF_DIR = RUNTIME_PATH / ".chalice"
+CHALICE_CONF_PATH = CHALICE_CONF_DIR / "config.json"
+CHALICE_CONF_BACKUP_PATH = CHALICE_CONF_DIR / "config.json.bak"
+
+
+def generate_chalice_config() -> None:
+    if not CHALICE_CONF_DIR.exists():
+        CHALICE_CONF_DIR.mkdir()
+
+    if CHALICE_CONF_PATH.exists():
+        CHALICE_CONF_PATH.rename(CHALICE_CONF_BACKUP_PATH)
+
+    CHALICE_CONF_PATH.write_text(
+        data=config_module.config.chalice_config.model_dump_json(
+            indent=2,
+            exclude_none=True,
+            by_alias=True,
+        )
+    )
+
+    if CHALICE_CONF_BACKUP_PATH.exists():
+        curr_file_md5 = hashlib.md5(string=CHALICE_CONF_PATH.read_bytes(), usedforsecurity=False).hexdigest()
+        backup_file_md5 = hashlib.md5(string=CHALICE_CONF_BACKUP_PATH.read_bytes(), usedforsecurity=False).hexdigest()
+
+        if curr_file_md5 == backup_file_md5:
+            print("No changes detected in chalice config.")
+            CHALICE_CONF_BACKUP_PATH.unlink()
+        else:
+            print("Chalice config updated.")
+    else:
+        print("Chalice config generated.")
+
+
+cli_funcs = [generate_chalice_config]

runtime/chalicelib/cli/generate_chalice_config.py

@@ -1,8 +1,10 @@
 import contextlib
+import json
 import logging
 import typing
 import urllib.parse
 
+import chalice.constants
 import firebase_admin
 import firebase_admin.credentials
 import httpx
@@ -22,6 +24,89 @@ def log_response(resp: httpx.Response) -> None:
     logger.info(f"RES [{req.method}]{req.url}<{resp.status_code}> {resp.read().decode(errors='ignore')=}")
 
 
+class ChaliceAPIGatewayCustomDomainConfig(pydantic.BaseModel):
+    domain_name: str
+    certificate_arn: str
+    tls_version: typing.Literal["TLS_1_0", "TLS_1_2"] = chalice.constants.DEFAULT_TLS_VERSION  # TLS_1_2
+    url_prefix: str | None = None
+    tags: dict[str, str] | None = None
+
+
+class ChaliceLambdaConfig(pydantic.BaseModel):
+    autogen_policy: bool = True
+    iam_policy_file: pydantic.FilePath | None = None
+    iam_role_arn: str | None = None
+    manage_iam_role: bool = True
+    subnet_ids: list[str] | None = None
+    security_group_ids: list[str] | None = None
+
+    lambda_memory_size: int | None = chalice.constants.DEFAULT_LAMBDA_MEMORY_SIZE
+    lambda_timeout: int | None = chalice.constants.DEFAULT_LAMBDA_TIMEOUT
+    layers: list[str] | None = None
+    log_retention_in_days: int | None = None
+    reserved_concurrency: int | None = None
+
+    environment_variables: dict[str, str] | None = None
+    tags: dict[str, str] | None = None
+
+    @pydantic.field_validator("lambda_memory_size", mode="before")
+    @classmethod
+    def validate_lambda_memory_size(cls, lambda_memory_size: int) -> int:
+        if lambda_memory_size % 64 != 0:
+            raise ValueError("lambda_memory_size must be a multiple of 64")
+
+        return lambda_memory_size
+
+    @pydantic.model_validator(mode="after")
+    def validate_model(self) -> typing.Self:
+        if not (self.autogen_policy or self.iam_policy_file):
+            raise ValueError("iam_policy_file must be set when autogen_policy is False")
+
+        if not (self.manage_iam_role or self.iam_role_arn):
+            raise ValueError("iam_role_arn must be set when manage_iam_role is False")
+
+        if bool(self.subnet_ids) != bool(self.security_group_ids):
+            raise ValueError("subnet_ids and security_group_ids must be set together")
+
+        return self
+
+
+class ChaliceStageConfig(ChaliceLambdaConfig, pydantic.BaseModel):
+    api_gateway_stage: str = chalice.constants.DEFAULT_APIGATEWAY_STAGE_NAME
+
+    api_gateway_custom_domain: ChaliceAPIGatewayCustomDomainConfig | None = None
+    websocket_api_custom_domain: ChaliceAPIGatewayCustomDomainConfig | None = None
+
+    automatic_layer: bool = False
+    minimum_compression_size: int | None = pydantic.Field(
+        default=None,
+        ge=chalice.constants.MIN_COMPRESSION_SIZE,
+        le=chalice.constants.MAX_COMPRESSION_SIZE,
+    )
+    xray: bool = False
+    lambda_functions: dict[str, ChaliceLambdaConfig] | None = None
+
+
+class ChaliceRootConfig(ChaliceStageConfig, pydantic.BaseModel):
+    api_gateway_endpoint_type: typing.Literal["EDGE", "REGIONAL", "PRIVATE"] = chalice.constants.DEFAULT_ENDPOINT_TYPE
+    api_gateway_endpoint_vpce: list[str] | str | None = None
+    api_gateway_policy_file: pydantic.FilePath | None = None
+
+    stages: dict[str, ChaliceStageConfig] | None = None
+
+    @pydantic.model_validator(mode="after")
+    def validate_model(self) -> typing.Self:
+        if self.api_gateway_endpoint_type == "PRIVATE" and not self.api_gateway_endpoint_vpce:
+            raise ValueError("api_gateway_endpoint_vpce must be set when api_gateway_endpoint_type is PRIVATE")
+
+        return self
+
+
+class ChaliceConfig(ChaliceRootConfig, pydantic.BaseModel):
+    version: typing.Literal["2.0"] = "2.0"
+    app_name: str
+
+
 class InfraConfig(pydantic_settings.BaseSettings):
     ecr_repo_name: str = "notico"
     lambda_name: str = "notico-lambda"
@@ -66,6 +151,10 @@ def get_session(self, service: AllowedToastServices) -> httpx.Client:  # type: i
             event_hooks={"request": [log_request], "response": [log_response]},
         )
 
+    @pydantic.field_serializer("secret_key", "sender_key", when_used="json-unless-none")
+    def dump_secret(self, v: pydantic.SecretStr) -> str:
+        return v.get_secret_value()
+
 
 class FirebaseConfig(ServiceConfig, pydantic_settings.BaseSettings):
     certificate: pydantic.SecretStr | None = None
@@ -81,6 +170,10 @@ def get_session(self) -> firebase_admin.App | None:  # type: ignore[override]
 
         return self._app
 
+    @pydantic.field_serializer("certificate", when_used="json-unless-none")
+    def dump_secret(self, v: pydantic.SecretStr) -> str:
+        return v.get_secret_value()
+
 
 class TelegramConfig(ServiceConfig, pydantic_settings.BaseSettings):
     bot_token: pydantic.SecretStr | None = None
@@ -93,20 +186,70 @@ def get_session(self) -> httpx.Client:
             event_hooks={"request": [log_request], "response": [log_response]},
         )
 
+    @pydantic.field_serializer("bot_token", when_used="json-unless-none")
+    def dump_secret(self, v: pydantic.SecretStr) -> str:
+        return v.get_secret_value()
+
 
 class SlackConfig(ServiceConfig, pydantic_settings.BaseSettings):
     channel: str | None = None
     token: pydantic.SecretStr | None = None
 
+    @pydantic.field_serializer("token", when_used="json-unless-none")
+    def dump_secret(self, v: pydantic.SecretStr) -> str:
+        return v.get_secret_value()
+
 
 class Config(pydantic_settings.BaseSettings):
+    project_name: str = "notico"
+
+    chalice: ChaliceRootConfig = pydantic.Field(default_factory=ChaliceRootConfig)
     infra: InfraConfig = pydantic.Field(default_factory=InfraConfig)
     toast: ToastConfig = pydantic.Field(default_factory=ToastConfig)
     firebase: FirebaseConfig = pydantic.Field(default_factory=FirebaseConfig)
     slack: SlackConfig = pydantic.Field(default_factory=SlackConfig)
     telegram: TelegramConfig = pydantic.Field(default_factory=TelegramConfig)
 
-    env_vars: dict[str, str] = pydantic.Field(default_factory=dict)
+    model_config = pydantic_settings.SettingsConfigDict(
+        env_nested_delimiter="__",
+        case_sensitive=False,
+        arbitrary_types_allowed=True,
+    )
+
+    @property
+    def env_dict(self) -> dict[str, str]:
+        settings_dict: dict[str, typing.Any] = self.model_dump(
+            mode="json",
+            by_alias=True,
+            exclude=["chalice"],
+            exclude_unset=True,
+            exclude_none=True,
+        )
+        result_dict: dict[str, str] = {}
+
+        if not (delimiters := self.model_config.get("env_nested_delimiter")):
+            raise ValueError("env_nested_delimiter must be set in model_config")
+
+        def _flatten_dict(d: dict, prefix: str = "") -> None:
+            for k, v in d.items():
+                key = f"{prefix}{delimiters}{k}" if prefix else k
+                if isinstance(v, dict):
+                    _flatten_dict(v, key)
+                elif isinstance(v, str):
+                    result_dict[key] = v
+                else:
+                    result_dict[key] = json.dumps(v)
+
+        _flatten_dict(settings_dict)
+        return result_dict
+
+    @property
+    def chalice_config(self) -> ChaliceConfig:
+        return ChaliceConfig(
+            app_name=self.project_name,
+            environment_variables=self.env_dict,
+            **self.chalice.model_dump(mode="python", exclude_unset=True, exclude_none=True),
+        )
 
 
-config = Config(_env_nested_delimiter="__", _case_sensitive=False)
+config = Config()