Added Model_mappings for eu region

Author: Maha Lakshmi Soumya Yalavarthi

src/lambda/update_configuration/index.py

@@ -21,7 +21,6 @@
 
 s3_client = boto3.client("s3")
 
-
 def fetch_content_from_s3(s3_uri: str) -> Union[Dict[str, Any], str]:
     """
     Fetches content from S3 URI and parses as JSON or YAML if possible
@@ -70,6 +69,23 @@ def resolve_content(content: Union[str, Dict[str, Any]]) -> Union[Dict[str, Any]
         return fetch_content_from_s3(content)
     return content
 
+# Model mapping between regions
+MODEL_MAPPINGS = {
+    "us.amazon.nova-lite-v1:0": "eu.amazon.nova-lite-v1:0",
+    "us.amazon.nova-pro-v1:0": "eu.amazon.nova-pro-v1:0",
+    "us.amazon.nova-premier-v1:0": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
+    "us.anthropic.claude-3-haiku-20240307-v1:0": "eu.anthropic.claude-3-haiku-20240307-v1:0",
+    "us.anthropic.claude-3-5-haiku-20241022-v1:0": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
+    "us.anthropic.claude-haiku-4-5-20251001-v1:0": "eu.anthropic.claude-haiku-4-5-20251001-v1:0",
+    "us.anthropic.claude-3-5-sonnet-20241022-v2:0": "eu.anthropic.claude-3-5-sonnet-20241022-v2:0",
+    "us.anthropic.claude-3-7-sonnet-20250219-v1:0": "eu.anthropic.claude-3-7-sonnet-20250219-v1:0",
+    "us.anthropic.claude-sonnet-4-20250514-v1:0": "eu.anthropic.claude-sonnet-4-20250514-v1:0",
+    "us.anthropic.claude-sonnet-4-5-20250929-v1:0:1m": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
+    "us.anthropic.claude-sonnet-4-5-20250929-v1:0": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
+    "us.anthropic.claude-sonnet-4-5-20250929-v1:0:1m": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0:1m",
+    "us.anthropic.claude-opus-4-20250514-v1:0": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
+    "us.anthropic.claude-opus-4-1-20250805-v1:0": "eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
+}
 
 def get_current_region() -> str:
     """Get the current AWS region"""
@@ -86,51 +102,83 @@ def is_us_region(region: str) -> bool:
     return region.startswith("us-")
 
 
-def _should_include_model(model_id: str, region_type: str) -> bool:
-    """Check if a model should be included for the given region type"""
-    # Non-string items are always included
-    if not isinstance(model_id, str):
-        return True
-
-    # For other regions or non-prefixed models, include everything
-    if region_type not in ["us", "eu"] or not (
-        model_id.startswith("us.") or model_id.startswith("eu.")
-    ):
-        return True
-
-    # For US regions: exclude EU models
-    if region_type == "us" and model_id.startswith("eu."):
-        return False
-
-    # For EU regions: exclude US models
-    if region_type == "eu" and model_id.startswith("us."):
-        return False
-
-    return True
+def get_model_mapping(model_id: str, target_region_type: str) -> str:
+    """Get the equivalent model for the target region type"""
+    if target_region_type == "eu":
+        return MODEL_MAPPINGS.get(model_id, model_id)
+    elif target_region_type == "us":
+        # Reverse mapping for US
+        for us_model, eu_model in MODEL_MAPPINGS.items():
+            if model_id == eu_model:
+                return us_model
+        return model_id
+    return model_id
 
 
 def filter_models_by_region(data: Any, region_type: str) -> Any:
     """Filter out models that don't match the region type"""
     if isinstance(data, dict):
-        return {
-            key: (
-                [item for item in value if _should_include_model(item, region_type)]
-                if isinstance(value, list)
-                and any(
-                    isinstance(item, str) and ("us." in item or "eu." in item)
-                    for item in value
-                )
-                else filter_models_by_region(value, region_type)
-            )
-            for key, value in data.items()
-        }
-
-    if isinstance(data, list):
+        filtered_data = {}
+        for key, value in data.items():
+            if isinstance(value, list) and any(
+                isinstance(item, str) and ("us." in item or "eu." in item)
+                for item in value
+            ):
+                # This is a model list - filter it
+                filtered_list = []
+                for item in value:
+                    if isinstance(item, str):
+                        # Include models that match the region type or are region-agnostic
+                        if region_type == "us":
+                            # Include US models and non-region-specific models, exclude EU models
+                            if item.startswith("us.") or (not item.startswith("eu.") and not item.startswith("us.")):
+                                filtered_list.append(item)
+                        elif region_type == "eu":
+                            # Include EU models and non-region-specific models, exclude US models
+                            if item.startswith("eu.") or (not item.startswith("eu.") and not item.startswith("us.")):
+                                filtered_list.append(item)
+                        else:
+                            # For other regions, include all models
+                            filtered_list.append(item)
+                    else:
+                        filtered_list.append(item)
+                filtered_data[key] = filtered_list
+            else:
+                filtered_data[key] = filter_models_by_region(value, region_type)
+        return filtered_data
+    elif isinstance(data, list):
         return [filter_models_by_region(item, region_type) for item in data]
+    return data
 
+
+def swap_model_ids(data: Any, region_type: str) -> Any:
+    """Swap model IDs to match the region type"""
+    if isinstance(data, dict):
+        swapped_data = {}
+        for key, value in data.items():
+            if isinstance(value, str) and ("us." in value or "eu." in value):
+                # This is a model ID - check if it needs swapping
+                if region_type == "us" and value.startswith("eu."):
+                    new_model = get_model_mapping(value, "us")
+                    if new_model != value:
+                        logger.info(f"Swapped EU model {value} to US model {new_model}")
+                    swapped_data[key] = new_model
+                elif region_type == "eu" and value.startswith("us."):
+                    new_model = get_model_mapping(value, "eu")
+                    if new_model != value:
+                        logger.info(f"Swapped US model {value} to EU model {new_model}")
+                    swapped_data[key] = new_model
+                else:
+                    swapped_data[key] = value
+            else:
+                swapped_data[key] = swap_model_ids(value, region_type)
+        return swapped_data
+    elif isinstance(data, list):
+        return [swap_model_ids(item, region_type) for item in data]
     return data
 
 
+
 def generate_physical_id(stack_id: str, logical_id: str) -> str:
     """
     Generates a consistent physical ID for the custom resource