feat: rework env & secrets

Author: Komalis

easyecs/cloudformation/template/task_definition.py

@@ -1,3 +1,7 @@
+import re
+from easyecs.model.ecs import EcsFileSecretModelV2
+
+
 def create_task_definition(
     stack, service_name, task_role, execution_role, log_group, ecs_data, run=False
 ):
@@ -116,11 +120,15 @@ def extract_container_config(stack, container_definition, log_configuration, run
     if tty and not run:
         command = ["sleep", "infinity"]
 
-    environment = {
-        env_definition.name: env_definition.value
-        for env_definition in container_definition.env
-        if env_definition.active
-    }
+    environment = {}
+    if isinstance(container_definition.env, List):
+        environment = {
+            env_definition.name: env_definition.value
+            for env_definition in container_definition.env
+            if env_definition.active
+        }
+    elif isinstance(container_definition.env, Dict):
+        environment = container_definition.env
 
     secrets = extract_secrets(stack, container_definition.secrets, name)
 
@@ -157,12 +165,27 @@ def extract_secrets(stack, secret_definitions, container_name):
     """Extract container secrets from its definition."""
     secrets = {}
     for secret_definition in secret_definitions:
-        secret_name = secret_definition.name
-        secret = Secret.from_secret_complete_arn(
-            stack, f"{secret_name}_{container_name}", secret_definition.arn
-        )
-        ecs_secret = ECSSecret.from_secrets_manager(secret, secret_definition.field)
-        secrets[secret_name] = ecs_secret
+        if isinstance(secret_definition, EcsFileSecretModel):
+            secret_name = secret_definition.name
+            secret = Secret.from_secret_complete_arn(
+                stack, f"{secret_name}_{container_name}", secret_definition.arn
+            )
+            ecs_secret = ECSSecret.from_secrets_manager(secret, secret_definition.field)
+            secrets[secret_name] = ecs_secret
+        elif isinstance(secret_definition, EcsFileSecretModelV2):
+            arn_fields = dict(re.finditer(r"^arn:aws:secretsmanager:(?P<region_name>[a-z0-9-]+):(?P<account_id>\d{12}):secret:(?P<secret_name>[^:]+)(?::(?P<field>[^:]*))?(?::([^:]*))?(?::([^:]*))?$", secret_definition.valueFrom))
+            if not arn_fields:
+                raise ValueError(f"Invalid ARN format: {secret_definition.valueFrom}")
+            field = arn_fields[0].group_dict()["field"]
+            import pdb; pdb.set_trace()
+            secret_name = secret_definition.name
+            secret = Secret.from_secret_complete_arn(
+                stack, f"{secret_name}_{container_name}", secret_definition.valueFrom
+            )
+            ecs_secret = ECSSecret.from_secrets_manager(secret)
+            secrets[secret_name] = ecs_secret
+        else:
+            raise Exception("Unsupported secret type")
     return secrets
 
 

easyecs/model/ecs.py

@@ -1,6 +1,6 @@
 import os
 import re
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 from pydantic import BaseModel, computed_field, field_validator, model_validator
 from pathlib import Path
 
@@ -71,6 +71,11 @@ class EcsFileSecretModel(BaseModel):
     active: bool = True
 
 
+class EcsFileSecretModelV2(BaseModel):
+    name: str
+    valueFrom: str
+
+
 class EcsFileVolumeModel(BaseModel):
     name: str
     id: str
@@ -101,8 +106,8 @@ class EcsFileContainerModel(BaseModel):
     resources: EcsFileResourcesModel
     build: Optional[EcsFileBuildModel] = None
     port_forward: List[str] = []
-    env: List[EcsFileEnvModel] = []
-    secrets: List[EcsFileSecretModel] = []
+    env: List[EcsFileEnvModel] | Dict[str, Any] | None = None
+    secrets: List[EcsFileSecretModel] | List[EcsFileSecretModelV2] = []
     efs_volumes: List[EcsFileVolumeModel] = []
     volumes: List[str] = []
     healthcheck: Optional[EcsFileContainerHealthCheckModel] = None