feat: allow explicit listing of function ids for local start-* 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-api' and '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/lib/local_api_service.py

@@ -40,7 +40,10 @@ def __init__(self, lambda_invoke_context, port, host, static_dir, disable_author
         self.cwd = lambda_invoke_context.get_cwd()
         self.disable_authorizer = disable_authorizer
         self.api_provider = ApiProvider(
-            lambda_invoke_context.stacks, cwd=self.cwd, disable_authorizer=disable_authorizer
+            lambda_invoke_context.stacks,
+            cwd=self.cwd,
+            disable_authorizer=disable_authorizer,
+            function_logical_ids=lambda_invoke_context.function_logical_ids,
         )
         self.lambda_runner = lambda_invoke_context.local_lambda_runner
         self.stderr_stream = lambda_invoke_context.stderr

samcli/commands/local/start_api/cli.py

@@ -60,6 +60,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(
@@ -107,6 +108,7 @@
 @print_cmdline_args
 def cli(
     ctx,
+    function_logical_ids,
     # start-api Specific Options
     host,
     port,
@@ -150,6 +152,7 @@ def cli(
 
     do_cli(
         ctx,
+        function_logical_ids,
         host,
         port,
         disable_authorizer,
@@ -183,6 +186,7 @@ def cli(
 
 def do_cli(  # pylint: disable=R0914
     ctx,
+    function_logical_ids,
     host,
     port,
     disable_authorizer,
@@ -256,6 +260,7 @@ def do_cli(  # pylint: disable=R0914
             container_host_interface=container_host_interface,
             invoke_images=processed_invoke_images,
             add_host=add_host,
+            function_logical_ids=function_logical_ids,
             no_mem_limit=no_mem_limit,
         ) as invoke_context:
             ssl_context = (ssl_cert_file, ssl_key_file) if ssl_cert_file else None

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/api_provider.py

@@ -1,19 +1,26 @@
 """Class that provides the Api with a list of routes from a Template"""
 
 import logging
-from typing import Iterator, List, Optional
+from typing import Iterator, List, Optional, Tuple
 
 from samcli.lib.providers.api_collector import ApiCollector
 from samcli.lib.providers.cfn_api_provider import CfnApiProvider
 from samcli.lib.providers.cfn_base_api_provider import CfnBaseApiProvider
 from samcli.lib.providers.provider import AbstractApiProvider, Api, Stack
 from samcli.lib.providers.sam_api_provider import SamApiProvider
+from samcli.local.apigw.route import Route
 
 LOG = logging.getLogger(__name__)
 
 
 class ApiProvider(AbstractApiProvider):
-    def __init__(self, stacks: List[Stack], cwd: Optional[str] = None, disable_authorizer: Optional[bool] = False):
+    def __init__(
+        self,
+        stacks: List[Stack],
+        cwd: Optional[str] = None,
+        disable_authorizer: Optional[bool] = False,
+        function_logical_ids: Optional[Tuple[str, ...]] = None,
+    ):
         """
         Initialize the class with template data. The template_dict is assumed
         to be valid, normalized and a dictionary. template_dict should be normalized by running any and all
@@ -31,13 +38,21 @@ def __init__(self, stacks: List[Stack], cwd: Optional[str] = None, disable_autho
             Optional working directory with respect to which we will resolve relative path to Swagger file
         disable_authorizer : Optional[bool]
             Optional flag to disable the collection of lambda authorizers
+        function_logical_ids : Optional[Tuple[str, ...]]
+            Optional tuple of function logical IDs to filter API routes by
         """
         self.stacks = stacks
 
         # Store a set of apis
         self.cwd = cwd
         self.disable_authorizer = disable_authorizer
+        self._function_logical_ids = function_logical_ids
         self.api = self._extract_api()
+
+        # Filter routes if function names provided
+        if function_logical_ids:
+            self.api = self._filter_api_routes(self.api, function_logical_ids)
+
         self.routes = self.api.routes
         LOG.debug("%d APIs found in the template", len(self.routes))
 
@@ -50,6 +65,50 @@ def get_all(self) -> Iterator[Api]:
 
         yield self.api
 
+    def _filter_api_routes(self, api: Api, function_logical_ids: Tuple[str, ...]) -> Api:
+        """
+        Filters API routes to only include those backed by specified functions.
+        Logs warnings for functions not referenced by any routes.
+
+        Parameters
+        ----------
+        api : Api
+            The Api object containing all routes
+        function_logical_ids : Tuple[str, ...]
+            Tuple of function logical IDs to filter by
+
+        Returns
+        -------
+        Api
+            The same Api object with filtered routes
+        """
+        function_logical_ids_set = set(function_logical_ids)
+
+        # Only filter if routes is a list (not a set of strings)
+        if not isinstance(api.routes, list):
+            return api
+
+        functions_with_routes = {
+            route.function_name
+            for route in api.routes
+            if isinstance(route, Route) and route.function_name in function_logical_ids_set
+        }
+        api.routes = [
+            route
+            for route in api.routes
+            if isinstance(route, Route) and route.function_name in function_logical_ids_set
+        ]
+
+        # Warn about functions without routes
+        functions_without_routes = function_logical_ids_set - functions_with_routes
+        if functions_without_routes:
+            LOG.warning(
+                "The following functions are not referenced by any API routes: %s",
+                ", ".join(sorted(functions_without_routes)),
+            )
+
+        return api
+
     def _extract_api(self) -> Api:
         """
         Extracts all the routes by running through the one providers. The provider that has the first type matched