Fix #5647: Add support for Fn::ForEach intrinsic function

samcli/commands/_utils/template.py

@@ -148,7 +148,11 @@ def _update_relative_paths(template_dict, original_root, new_root):
 
             properties[path_prop_name] = updated_path
 
-    for _, resource in template_dict.get("Resources", {}).items():
+    for resource_id, resource in template_dict.get("Resources", {}).items():
+        # Skip Fn::ForEach constructs which are lists, not dicts
+        if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+            continue
+
         resource_type = resource.get("Type")
 
         if resource_type not in RESOURCES_WITH_LOCAL_PATHS:

samcli/lib/iac/cdk/utils.py

@@ -42,7 +42,10 @@ def _resource_level_metadata_exists(resources: Dict) -> bool:
         Dict of resources to look through
 
     """
-    for _, resource in resources.items():
+    for resource_id, resource in resources.items():
+        # Skip Fn::ForEach constructs which are lists, not dicts
+        if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+            continue
         if resource.get("Type", "") == CDK_METADATA_TYPE_VALUE:
             return True
     return False
@@ -58,7 +61,10 @@ def _cdk_path_metadata_exists(resources: Dict) -> bool:
         Dict of resources to look through
 
     """
-    for _, resource in resources.items():
+    for resource_id, resource in resources.items():
+        # Skip Fn::ForEach constructs which are lists, not dicts
+        if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+            continue
         metadata = resource.get("Metadata", {})
         if metadata and CDK_PATH_METADATA_KEY in metadata:
             return True

samcli/lib/providers/sam_function_provider.py

@@ -202,6 +202,11 @@ def _extract_functions(
         result: Dict[str, Function] = {}  # a dict with full_path as key and extracted function as value
         for stack in stacks:
             for name, resource in stack.resources.items():
+                # Skip Fn::ForEach constructs which are lists, not dicts
+                if name.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+                    LOG.debug(f"Skipping Fn::ForEach construct or non-dict resource: {name}")
+                    continue
+
                 resource_type = resource.get("Type")
                 resource_properties = resource.get("Properties", {})
                 resource_metadata = resource.get("Metadata", None)

samcli/lib/providers/sam_layer_provider.py

@@ -83,6 +83,11 @@ def _extract_layers(self) -> List[LayerVersion]:
         layers = []
         for stack in self._stacks:
             for name, resource in stack.resources.items():
+                # Skip Fn::ForEach constructs which are lists, not dicts
+                if name.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+                    LOG.debug(f"Skipping Fn::ForEach construct or non-dict resource: {name}")
+                    continue
+
                 # In the list of layers that is defined within a template, you can reference a LayerVersion resource.
                 # When running locally, we need to follow that Ref so we can extract the local path to the layer code.
                 resource_type = resource.get("Type")

samcli/lib/providers/sam_stack_provider.py

@@ -110,6 +110,11 @@ def _extract_stacks(self) -> None:
         """
 
         for name, resource in self._resources.items():
+            # Skip Fn::ForEach constructs which are lists, not dicts
+            if name.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+                LOG.debug(f"Skipping Fn::ForEach construct or non-dict resource: {name}")
+                continue
+
             resource_type = resource.get("Type")
             resource_properties = resource.get("Properties", {})
             resource_metadata = resource.get("Metadata", None)

samcli/lib/samlib/resource_metadata_normalizer.py

@@ -62,6 +62,10 @@ def normalize(template_dict, normalize_parameters=False):
         resources = template_dict.get(RESOURCES_KEY, {})
 
         for logical_id, resource in resources.items():
+            # Skip Fn::ForEach constructs which are lists, not dicts
+            if logical_id.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+                continue
+
             # copy metadata to another variable, change its values and assign it back in the end
             resource_metadata = deepcopy(resource.get(METADATA_KEY)) or {}
 
@@ -228,6 +232,10 @@ def get_resource_id(resource_properties, logical_id):
         str
             The unique function id
         """
+        # Skip Fn::ForEach constructs which are lists, not dicts
+        if not isinstance(resource_properties, dict):
+            return logical_id
+
         resource_metadata = resource_properties.get("Metadata", {})
         customer_defined_id = resource_metadata.get(SAM_RESOURCE_ID_KEY)
 

samcli/lib/samlib/wrapper.py

@@ -26,6 +26,7 @@
 from samtranslator.validator.validator import SamTemplateValidator
 
 from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException
+from samcli.lib.utils.foreach_handler import filter_foreach_constructs
 
 from .local_uri_plugin import SupportLocalUriPlugin
 
@@ -69,8 +70,19 @@ def run_plugins(self, convert_local_uris=True):
         # Temporarily disabling validation for DeletionPolicy and UpdateReplacePolicy when language extensions are set
         self._patch_language_extensions()
 
+        # Filter out Fn::ForEach constructs before parsing
+        # CloudFormation will handle these server-side
+        template_copy, foreach_constructs = filter_foreach_constructs(template_copy)
+
         try:
             parser.parse(template_copy, all_plugins)  # parse() will run all configured plugins
+
+            # Add back Fn::ForEach constructs after parsing
+            if foreach_constructs:
+                if "Resources" not in template_copy:
+                    template_copy["Resources"] = {}
+                template_copy["Resources"].update(foreach_constructs)
+
         except InvalidDocumentException as e:
             raise InvalidSamDocumentException(
                 functools.reduce(lambda message, error: message + " " + str(error), e.causes, str(e))
@@ -100,6 +112,8 @@ def patched_func(self):
 
             SamResource.valid = patched_func
 
+    # Removed: _filter_foreach_constructs() now uses shared utility from samcli.lib.utils.foreach_handler
+
     @staticmethod
     def _check_using_language_extension(template: Dict) -> bool:
         """

samcli/lib/translate/sam_template_validator.py

@@ -13,6 +13,7 @@
 from samtranslator.translator.translator import Translator
 
 from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException
+from samcli.lib.utils.foreach_handler import filter_foreach_constructs
 from samcli.lib.utils.packagetype import IMAGE, ZIP
 from samcli.lib.utils.resources import AWS_SERVERLESS_FUNCTION
 from samcli.yamlhelper import yaml_dump
@@ -82,19 +83,33 @@ def get_translated_template_if_valid(self):
         self._replace_local_codeuri()
         self._replace_local_image()
 
+        # Filter out Fn::ForEach constructs before translation
+        # CloudFormation will handle these server-side
+        template_to_translate, foreach_constructs = filter_foreach_constructs(self.sam_template)
+
         try:
             template = sam_translator.translate(
-                sam_template=self.sam_template,
+                sam_template=template_to_translate,
                 parameter_values=self.parameter_overrides,
                 get_managed_policy_map=self._get_managed_policy_map,
             )
+
+            # Add back Fn::ForEach constructs after translation
+            if foreach_constructs:
+                if "Resources" not in template:
+                    template["Resources"] = {}
+                template["Resources"].update(foreach_constructs)
+                LOG.debug("Preserved %d Fn::ForEach construct(s) in template", len(foreach_constructs))
+
             LOG.debug("Translated template is:\n%s", yaml_dump(template))
             return yaml_dump(template)
         except InvalidDocumentException as e:
             raise InvalidSamDocumentException(
                 functools.reduce(lambda message, error: message + " " + str(error), e.causes, str(e))
             ) from e
 
+    # Removed: _filter_foreach_constructs() now uses shared utility from samcli.lib.utils.foreach_handler
+
     @functools.lru_cache(maxsize=None)
     def _get_managed_policy_map(self) -> Dict[str, str]:
         """
@@ -130,7 +145,11 @@ def _replace_local_codeuri(self):
                 ):
                     SamTemplateValidator._update_to_s3_uri("CodeUri", properties)
 
-        for _, resource in all_resources.items():
+        for resource_id, resource in all_resources.items():
+            # Skip Fn::ForEach constructs which are lists, not dicts
+            if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+                continue
+
             resource_type = resource.get("Type")
             resource_dict = resource.get("Properties", {})
 
@@ -158,7 +177,11 @@ def _replace_local_image(self):
         This ensures sam validate works without having to package the app or use ImageUri.
         """
         resources = self.sam_template.get("Resources", {})
-        for _, resource in resources.items():
+        for resource_id, resource in resources.items():
+            # Skip Fn::ForEach constructs which are lists, not dicts
+            if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict):
+                continue
+
             resource_type = resource.get("Type")
             properties = resource.get("Properties", {})
 

samcli/lib/utils/foreach_handler.py

@@ -0,0 +1,66 @@
+"""
+Utility functions for handling CloudFormation Fn::ForEach intrinsic function
+"""
+import copy
+import logging
+from typing import Dict, Tuple
+
+LOG = logging.getLogger(__name__)
+
+
+def filter_foreach_constructs(template: Dict) -> Tuple[Dict, Dict]:
+    """
+    Filter out Fn::ForEach constructs from template before SAM transformation.
+    CloudFormation will handle these server-side during deployment.
+
+    Parameters
+    ----------
+    template : Dict
+        The SAM/CloudFormation template dictionary
+
+    Returns
+    -------
+    Tuple[Dict, Dict]
+        (template_without_foreach, foreach_constructs_dict)
+
+    Notes
+    -----
+    Fn::ForEach constructs are identified by resource IDs starting with "Fn::ForEach::"
+    These constructs are lists, not dicts, and would cause parsing errors if processed locally.
+    CloudFormation expands them server-side during deployment.
+    """
+    template_copy = copy.deepcopy(template)
+    resources = template_copy.get("Resources", {})
+
+    # If no Resources section, nothing to filter
+    if not resources:
+        return template_copy, {}
+
+    # Separate Fn::ForEach constructs from regular resources
+    foreach_constructs = {}
+    regular_resources = {}
+
+    for resource_id, resource in resources.items():
+        if resource_id.startswith("Fn::ForEach::"):
+            foreach_constructs[resource_id] = resource
+            LOG.info(
+                f"Detected Fn::ForEach construct '{resource_id}'. "
+                "This will be expanded by CloudFormation during deployment."
+            )
+        else:
+            regular_resources[resource_id] = resource
+
+    # If template only has ForEach constructs, add a placeholder resource
+    # to satisfy SAM Translator's requirement for non-empty Resources section
+    if not regular_resources and foreach_constructs:
+        regular_resources["__PlaceholderForForEachOnly"] = {
+            "Type": "AWS::CloudFormation::WaitConditionHandle",
+            "Properties": {},
+        }
+        LOG.debug("Added placeholder resource since template only contains Fn::ForEach constructs")
+
+    # Only update Resources if there were any ForEach constructs
+    if foreach_constructs:
+        template_copy["Resources"] = regular_resources
+
+    return template_copy, foreach_constructs

tests/integration/testdata/fn-foreach/basic-topics-template.yaml

@@ -0,0 +1,14 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform:
+  - AWS::LanguageExtensions
+  - AWS::Serverless-2016-10-31
+
+Resources:
+  'Fn::ForEach::Topics':
+    - TopicName
+    - [Success, Failure, Timeout, Unknown]
+    - 'SnsTopic${TopicName}':
+        Type: AWS::SNS::Topic
+        Properties:
+          TopicName: !Ref TopicName
+          FifoTopic: true