feat: allow explicit listing of function ids for local start-lambda command

Author: vicheey

samcli/commands/local/cli_common/invoke_context.py

@@ -98,6 +98,7 @@ def __init__(
         invoke_images: Optional[str] = None,
         mount_symlinks: Optional[bool] = False,
         no_mem_limit: Optional[bool] = False,
+        function_logical_ids: Optional[Tuple[str, ...]] = None,
     ) -> None:
         """
         Initialize the context
@@ -107,7 +108,7 @@ def __init__(
         template_file str
             Name or path to template
         function_identifier str
-            Identifier of the function to invoke
+            Identifier of the single function to invoke (used by 'sam local invoke' command)
         env_vars_file str
             Path to a file containing values for environment variables
         docker_volume_basedir str
@@ -154,10 +155,27 @@ def __init__(
             Optional. A dictionary that defines the custom invoke image URI of each function
         mount_symlinks bool
             Optional. Indicates if symlinks should be mounted inside the container
+        function_logical_ids tuple(str)
+            Optional. Tuple of function logical IDs to filter and make available for local execution.
+            Used by 'sam local start-api' and 'sam local start-lambda' commands to limit which
+            functions from the template are exposed. If not provided, all functions are available.
         """
 
         self._template_file = template_file
         self._function_identifier = function_identifier
+        self._function_logical_ids = function_logical_ids
+
+        # Validate that function_identifier and function_logical_ids aren't both provided
+        # function_identifier is for 'sam local invoke' (single function)
+        # function_logical_ids is for 'sam local start-*' commands (multiple function filter)
+        if self._function_identifier and self._function_logical_ids:
+            LOG.warning(
+                "Both function_identifier and function_logical_ids were provided. "
+                "function_identifier is used for 'sam local invoke' to specify a single function, "
+                "while function_logical_ids is used for 'sam local start-api/start-lambda' to filter functions. "
+                "function_identifier will take precedence for single function invocation."
+            )
+
         self._env_vars_file = env_vars_file
         self._docker_volume_basedir = docker_volume_basedir
         self._docker_network = docker_network
@@ -241,10 +259,18 @@ def __enter__(self) -> "InvokeContext":
         if self._docker_volume_basedir:
             _function_providers_args[self._containers_mode].append(True)
 
+        # Add function_logical_ids parameter to both provider types
+        _function_providers_kwargs: Dict[str, Any] = {}
+        if self._function_logical_ids is not None:
+            _function_providers_kwargs["function_logical_ids"] = self._function_logical_ids
+
         self._function_provider = _function_providers_class[self._containers_mode](
-            *_function_providers_args[self._containers_mode]
+            *_function_providers_args[self._containers_mode], **_function_providers_kwargs
         )
 
+        # Validate function logical IDs after provider is initialized
+        self._validate_function_logical_ids()
+
         self._env_vars_value = self._get_env_vars_value(self._env_vars_file)
         self._container_env_vars_value = self._get_env_vars_value(self._container_env_vars_file)
         self._log_file_handle = self._setup_log_file(self._log_file)
@@ -448,6 +474,34 @@ def _clean_running_containers_and_related_resources(self) -> None:
         cast(WarmLambdaRuntime, self.lambda_runtime).clean_running_containers_and_related_resources()
         cast(RefreshableSamFunctionProvider, self._function_provider).stop_observer()
 
+    def _validate_function_logical_ids(self) -> None:
+        """
+        Validates that all provided function logical IDs exist in the template.
+        Raises InvalidFunctionNamesException with helpful error message if validation fails.
+        """
+        if not self._function_logical_ids:
+            return  # No filtering requested
+
+        # Get all available function names/IDs from the provider
+        all_functions_set = set()
+        for func in self._function_provider.get_all():
+            all_functions_set.add(func.name)
+            all_functions_set.add(func.function_id)
+            if func.functionname:
+                all_functions_set.add(func.functionname)
+
+        # Check for invalid IDs
+        invalid_ids = set(self._function_logical_ids) - all_functions_set
+
+        if invalid_ids:
+            available_ids = sorted(all_functions_set)
+            from samcli.commands.local.cli_common.user_exceptions import InvalidFunctionNamesException
+
+            raise InvalidFunctionNamesException(
+                f"Invalid function logical ID(s): {', '.join(sorted(invalid_ids))}\n\n"
+                f"Available functions in template:\n  " + "\n  ".join(available_ids)
+            )
+
     def _add_account_id_to_global(self) -> None:
         """
         Attempts to get the Account ID from the current session
@@ -556,6 +610,21 @@ def initialized_container_ids(self) -> List[str]:
         """
         return self._initialized_container_ids
 
+    @property
+    def function_logical_ids(self) -> Optional[Tuple[str, ...]]:
+        """
+        Returns the tuple of function logical IDs to filter, if provided.
+
+        This is used by 'sam local start-lambda' commands
+        to limit which functions from the template are made available for local execution.
+        This is different from function_identifier which is used by 'sam local invoke'
+        to specify a single function to invoke.
+
+        Returns:
+            Optional[Tuple[str, ...]]: Tuple of function logical IDs or None if no filter
+        """
+        return self._function_logical_ids
+
     @property
     def stdout(self) -> StreamWriter:
         """

samcli/commands/local/cli_common/user_exceptions.py

@@ -11,6 +11,12 @@ class InvokeContextException(UserException):
     """
 
 
+class InvalidFunctionNamesException(InvokeContextException):
+    """
+    User provided function names that don't exist in the template
+    """
+
+
 class InvalidSamTemplateException(UserException):
     """
     The template provided was invalid and not able to transform into a Standard CloudFormation Template

samcli/commands/local/start_lambda/cli.py

@@ -52,6 +52,7 @@
     requires_credentials=False,
     context_settings={"max_content_width": 120},
 )
+@click.argument("function_logical_ids", nargs=-1, required=False)
 @configuration_option(provider=ConfigProvider(section="parameters"))
 @terraform_plan_file_option
 @hook_name_click_option(
@@ -71,6 +72,7 @@
 @print_cmdline_args
 def cli(
     ctx,  # pylint: disable=R0914
+    function_logical_ids,
     # start-lambda Specific Options
     host,
     port,
@@ -110,6 +112,7 @@ def cli(
 
     do_cli(
         ctx,
+        function_logical_ids,
         host,
         port,
         template_file,
@@ -139,6 +142,7 @@ def cli(
 
 def do_cli(  # pylint: disable=R0914
     ctx,
+    function_logical_ids,
     host,
     port,
     template,
@@ -208,6 +212,7 @@ def do_cli(  # pylint: disable=R0914
             container_host_interface=container_host_interface,
             add_host=add_host,
             invoke_images=processed_invoke_images,
+            function_logical_ids=function_logical_ids,
             no_mem_limit=no_mem_limit,
         ) as invoke_context:
             service = LocalLambdaService(lambda_invoke_context=invoke_context, port=port, host=host)

samcli/lib/providers/sam_function_provider.py

@@ -4,7 +4,7 @@
 
 import logging
 from pathlib import Path
-from typing import Any, Dict, Iterator, List, Optional, cast
+from typing import Any, Dict, Iterator, List, Optional, Tuple, cast
 
 from samtranslator.policy_template_processor.exceptions import TemplateNotFoundException
 
@@ -45,6 +45,7 @@ def __init__(
         use_raw_codeuri: bool = False,
         ignore_code_extraction_warnings: bool = False,
         locate_layer_nested: bool = False,
+        function_logical_ids: Optional[Tuple[str, ...]] = None,
     ) -> None:
         """
         Initialize the class with SAM template data. The SAM template passed to this provider is assumed
@@ -61,16 +62,18 @@ def __init__(
             Note(xinhol): use_raw_codeuri is temporary to fix a bug, and will be removed for a permanent solution.
         :param bool ignore_code_extraction_warnings: Ignores Log warnings
         :param bool locate_layer_nested: resolved nested layer reference to their actual location in the nested stack
+        :param tuple function_logical_ids: Optional tuple of function logical IDs to filter by
         """
 
         self._stacks = stacks
+        self._function_logical_ids = function_logical_ids
 
         for stack in stacks:
             LOG.debug("%d resources found in the stack %s", len(stack.resources), stack.stack_path)
 
         # Store a map of function full_path to function information for quick reference
         self.functions = SamFunctionProvider._extract_functions(
-            self._stacks, use_raw_codeuri, ignore_code_extraction_warnings, locate_layer_nested
+            self._stacks, use_raw_codeuri, ignore_code_extraction_warnings, locate_layer_nested, function_logical_ids
         )
 
         self._colored = Colored()
@@ -90,6 +93,7 @@ def update(
         use_raw_codeuri: bool = False,
         ignore_code_extraction_warnings: bool = False,
         locate_layer_nested: bool = False,
+        function_logical_ids: Optional[Tuple[str, ...]] = None,
     ) -> None:
         """
         Hydrate the function provider with updated stacks
@@ -98,10 +102,11 @@ def update(
             Note(xinhol): use_raw_codeuri is temporary to fix a bug, and will be removed for a permanent solution.
         :param bool ignore_code_extraction_warnings: Ignores Log warnings
         :param bool locate_layer_nested: resolved nested layer reference to their actual location in the nested stack
+        :param tuple function_logical_ids: Optional tuple of function logical IDs to filter by
         """
         self._stacks = stacks
         self.functions = SamFunctionProvider._extract_functions(
-            self._stacks, use_raw_codeuri, ignore_code_extraction_warnings, locate_layer_nested
+            self._stacks, use_raw_codeuri, ignore_code_extraction_warnings, locate_layer_nested, function_logical_ids
         )
 
     def get(self, name: str) -> Optional[Function]:
@@ -180,12 +185,29 @@ def get_all(self) -> Iterator[Function]:
         for _, function in self.functions.items():
             yield function
 
+    @staticmethod
+    def _should_include_function(function: Function, function_logical_ids: set) -> bool:
+        """
+        Determines if a function should be included based on the filter.
+        Matches against function_id, name, and functionname.
+
+        :param function: Function object to check
+        :param function_logical_ids: Set of function logical IDs to match against
+        :return bool: True if function should be included, False otherwise
+        """
+        return bool(
+            function.function_id in function_logical_ids
+            or function.name in function_logical_ids
+            or (function.functionname and function.functionname in function_logical_ids)
+        )
+
     @staticmethod
     def _extract_functions(
         stacks: List[Stack],
         use_raw_codeuri: bool = False,
         ignore_code_extraction_warnings: bool = False,
         locate_layer_nested: bool = False,
+        function_logical_ids: Optional[Tuple[str, ...]] = None,
     ) -> Dict[str, Function]:
         """
         Extracts and returns function information from the given dictionary of SAM/CloudFormation resources. This
@@ -195,6 +217,7 @@ def _extract_functions(
         :param bool use_raw_codeuri: Do not resolve adjust core_uri based on the template path, use the raw uri.
         :param bool ignore_code_extraction_warnings: suppress log statements on code extraction from resources.
         :param bool locate_layer_nested: resolved nested layer reference to their actual location in the nested stack
+        :param tuple function_logical_ids: Optional tuple of function logical IDs to filter by
         :return dict(string : samcli.commands.local.lib.provider.Function): Dictionary of function full_path to the
             Function configuration object
         """
@@ -271,6 +294,17 @@ def _extract_functions(
 
                 # We don't care about other resource types. Just ignore them
 
+        # Apply filtering if function_logical_ids provided
+        if function_logical_ids:
+            function_logical_ids_set = set(function_logical_ids)
+            filtered_functions = {}
+
+            for full_path, function in result.items():
+                if SamFunctionProvider._should_include_function(function, function_logical_ids_set):
+                    filtered_functions[full_path] = function
+
+            return filtered_functions
+
         return result
 
     @staticmethod
@@ -800,6 +834,7 @@ def __init__(
         global_parameter_overrides: Optional[Dict] = None,
         use_raw_codeuri: bool = False,
         ignore_code_extraction_warnings: bool = False,
+        function_logical_ids: Optional[Tuple[str, ...]] = None,
     ) -> None:
         """
         Initialize the class with SAM template data. The SAM template passed to this provider is assumed
@@ -815,9 +850,15 @@ def __init__(
         :param bool use_raw_codeuri: Do not resolve adjust core_uri based on the template path, use the raw uri.
             Note(xinhol): use_raw_codeuri is temporary to fix a bug, and will be removed for a permanent solution.
         :param bool ignore_code_extraction_warnings: Ignores Log warnings
+        :param tuple function_logical_ids: Optional tuple of function logical IDs to filter by
         """
 
-        super().__init__(stacks, use_raw_codeuri, ignore_code_extraction_warnings)
+        # Store function_logical_ids before calling super().__init__
+        self._function_logical_ids = function_logical_ids
+
+        super().__init__(
+            stacks, use_raw_codeuri, ignore_code_extraction_warnings, function_logical_ids=function_logical_ids
+        )
 
         # initialize root templates. Usually it will be one template, as sam commands only support processing
         # one template. These templates will be fixed.
@@ -909,6 +950,7 @@ def _watch_stack_templates(self, stacks: List[Stack]) -> None:
     def _refresh_loaded_functions(self) -> None:
         """
         Reload the stacks, and lambda functions from template files.
+        Applies the same function filter during refresh.
         """
         LOG.debug("A change got detected in one of the stack templates. Reload the lambda function resources")
         self._stacks = []
@@ -925,8 +967,12 @@ def _refresh_loaded_functions(self) -> None:
                 raise ex
 
         self.is_changed = False
+        # Pass function_logical_ids to maintain filter during refresh
         self.functions = self._extract_functions(
-            self._stacks, self._use_raw_codeuri, self._ignore_code_extraction_warnings
+            self._stacks,
+            self._use_raw_codeuri,
+            self._ignore_code_extraction_warnings,
+            function_logical_ids=self._function_logical_ids,
         )
         self._watch_stack_templates(self._stacks)
 

tests/integration/local/start_lambda/start_lambda_api_integ_base.py

@@ -37,6 +37,7 @@ class StartLambdaIntegBaseClass(TestCase):
     terraform_plan_file: Optional[str] = None
     beta_features: Optional[bool] = None
     collect_start_lambda_process_output: bool = False
+    function_logical_ids: Optional[List[str]] = None
 
     build_before_invoke = False
     build_overrides: Optional[Dict[str, str]] = None
@@ -114,9 +115,14 @@ def get_start_lambda_command(
         hook_name=None,
         beta_features=None,
         terraform_plan_file=None,
+        function_logical_ids=None,
     ):
         command_list = [get_sam_command(), "local", "start-lambda"]
 
+        # Add function names as positional arguments first
+        if function_logical_ids:
+            command_list.extend(function_logical_ids)
+
         if port:
             command_list += ["-p", port]
 
@@ -159,6 +165,7 @@ def start_lambda(cls, wait_time=5, input=None, env=None):
             hook_name=cls.hook_name,
             beta_features=cls.beta_features,
             terraform_plan_file=cls.terraform_plan_file,
+            function_logical_ids=cls.function_logical_ids,
         )
 
         # Container labels are no longer needed - container IDs are parsed from output