feat: support for local Event invocation type

samcli/commands/local/lib/local_lambda.py

@@ -96,47 +96,35 @@ def __init__(
         self.container_host_interface = container_host_interface
         self.extra_hosts = extra_hosts
 
-    def invoke(
-        self,
-        function_identifier: str,
-        event: str,
-        tenant_id: Optional[str] = None,
-        stdout: Optional[StreamWriter] = None,
-        stderr: Optional[StreamWriter] = None,
-        override_runtime: Optional[str] = None,
-        invocation_type: str = "RequestResponse",
-        durable_execution_name: Optional[str] = None,
-    ) -> Optional[Dict[str, str]]:
+    def get_function(self, function_identifier: str, tenant_id: Optional[str] = None) -> Function:
         """
-        Find the Lambda function with given name and invoke it. Pass the given event to the function and return
-        response through the given streams.
-
-        This function will block until either the function completes or times out.
+        Get a Lambda function by identifier, raising FunctionNotFound if not found.
 
         Parameters
         ----------
-        function_identifier str
-            Identifier of the Lambda function to invoke, it can be logicalID, function name or full path
-        event str
-            Event data passed to the function. Must be a valid JSON String.
-        stdout samcli.lib.utils.stream_writer.StreamWriter
-            Stream writer to write the output of the Lambda function to.
-        stderr samcli.lib.utils.stream_writer.StreamWriter
-            Stream writer to write the Lambda runtime logs to.
-        Runtime: str
-            To use instead of the runtime specified in the function configuration
-        durable_execution_name: str
-            Optional name for the durable execution (for durable functions only)
-
+        function_identifier : str
+            Identifier of the Lambda function, it can be logicalID, function name or full path
+        tenant_id : Optional[str]
+            Optional tenant ID for multi-tenant functions
         Returns
         -------
-        Optional[Dict[str, str]]
-            HTTP headers dict if this was a durable function invocation, None otherwise
+        Function
+            The Lambda function configuration
 
         Raises
         ------
-        FunctionNotfound
-            When we cannot find a function with the given name
+        InvalidFunctionNameException
+            When the function identifier doesn't match AWS Lambda's validation pattern
+        FunctionNotFound
+            When we cannot find a function with the given identifier
+        TenantIdValidationError
+            When the tenant ID is not provided for multi-tenant functions
+        UnsupportedInlineCodeError
+            When the function has inline code and is being invoked locally
+        InvalidIntermediateImageError
+            When the function has an intermediate image and is being invoked locally
+        UnsupportedRuntimeArchitectureError
+            When the function runtime and architecture are not compatible
         """
         # Normalize function identifier from ARN if provided
         normalized_function_identifier = normalize_sam_function_identifier(function_identifier)
@@ -182,6 +170,55 @@ def invoke(
                 "Remove the tenant ID from your request and try again."
             )
 
+        return function
+
+    def invoke(
+        self,
+        function_identifier: str,
+        event: str,
+        tenant_id: Optional[str] = None,
+        stdout: Optional[StreamWriter] = None,
+        stderr: Optional[StreamWriter] = None,
+        override_runtime: Optional[str] = None,
+        invocation_type: str = "RequestResponse",
+        durable_execution_name: Optional[str] = None,
+        function: Optional[Function] = None,
+    ) -> Optional[Dict[str, str]]:
+        """
+        Find the Lambda function with given name and invoke it. Pass the given event to the function and return
+        response through the given streams.
+
+        This function will block until either the function completes or times out.
+
+        Parameters
+        ----------
+        function_identifier str
+            Identifier of the Lambda function to invoke, it can be logicalID, function name or full path
+        event str
+            Event data passed to the function. Must be a valid JSON String.
+        stdout samcli.lib.utils.stream_writer.StreamWriter
+            Stream writer to write the output of the Lambda function to.
+        stderr samcli.lib.utils.stream_writer.StreamWriter
+            Stream writer to write the Lambda runtime logs to.
+        Runtime: str
+            To use instead of the runtime specified in the function configuration
+        durable_execution_name: str
+            Optional name for the durable execution (for durable functions only)
+
+        Returns
+        -------
+        Optional[Dict[str, str]]
+            HTTP headers dict if this was a durable function invocation, None otherwise
+
+        Raises
+        ------
+        FunctionNotfound
+            When we cannot find a function with the given name
+        """
+        # Get the function configuration
+        if not function:
+            function = self.get_function(function_identifier, tenant_id)
+
         config = self.get_invoke_config(function, override_runtime)
 
         if (

samcli/local/lambda_service/lambda_error_responses.py

@@ -2,6 +2,9 @@
 
 import json
 from collections import OrderedDict
+from typing import Any, Dict
+
+from flask import Response
 
 from samcli.local.services.base_local_service import BaseLocalService
 
@@ -40,7 +43,7 @@ class LambdaErrorResponses:
     CONTENT_TYPE_HEADER_KEY = "Content-Type"
 
     @staticmethod
-    def resource_not_found(function_name):
+    def resource_not_found(function_name: str) -> Response:
         """
         Creates a Lambda Service ResourceNotFound Response
 
@@ -66,7 +69,7 @@ def resource_not_found(function_name):
         )
 
     @staticmethod
-    def invalid_request_content(message):
+    def invalid_request_content(message: str) -> Response:
         """
         Creates a Lambda Service InvalidRequestContent Response
 
@@ -89,7 +92,7 @@ def invalid_request_content(message):
         )
 
     @staticmethod
-    def validation_exception(message):
+    def validation_exception(message: str) -> Response:
         """
         Creates a Lambda Service ValidationException Response
 
@@ -112,7 +115,7 @@ def validation_exception(message):
         )
 
     @staticmethod
-    def unsupported_media_type(content_type):
+    def unsupported_media_type(content_type: str) -> Response:
         """
         Creates a Lambda Service UnsupportedMediaType Response
 
@@ -137,7 +140,7 @@ def unsupported_media_type(content_type):
         )
 
     @staticmethod
-    def generic_service_exception(*args):
+    def generic_service_exception(*args: Any) -> Response:
         """
         Creates a Lambda Service Generic ServiceException Response
 
@@ -160,7 +163,7 @@ def generic_service_exception(*args):
         )
 
     @staticmethod
-    def not_implemented_locally(message):
+    def not_implemented_locally(message: str) -> Response:
         """
         Creates a Lambda Service NotImplementedLocally Response
 
@@ -183,7 +186,7 @@ def not_implemented_locally(message):
         )
 
     @staticmethod
-    def generic_path_not_found(*args):
+    def generic_path_not_found(*args: Any) -> Response:
         """
         Creates a Lambda Service Generic PathNotFound Response
 
@@ -208,7 +211,7 @@ def generic_path_not_found(*args):
         )
 
     @staticmethod
-    def generic_method_not_allowed(*args):
+    def generic_method_not_allowed(*args: Any) -> Response:
         """
         Creates a Lambda Service Generic MethodNotAllowed Response
 
@@ -233,13 +236,13 @@ def generic_method_not_allowed(*args):
         )
 
     @staticmethod
-    def container_creation_failed(message):
+    def container_creation_failed(message: str) -> Response:
         """
         Creates a Container Creation Failed response
         Parameters
         ----------
-        args list
-            List of arguments Flask passes to the method
+        message str
+            Message to be added to the body of the response
         Returns
         -------
         Flask.Response
@@ -256,7 +259,7 @@ def container_creation_failed(message):
         )
 
     @staticmethod
-    def _construct_error_response_body(error_type, error_message):
+    def _construct_error_response_body(error_type: str, error_message: str) -> str:
         """
         Constructs a string to be used in the body of the Response that conforms
         to the structure of the Lambda Service Responses
@@ -278,7 +281,7 @@ def _construct_error_response_body(error_type, error_message):
 
     # Durable Functions Error Responses
     @staticmethod
-    def durable_execution_not_found(execution_arn):
+    def durable_execution_not_found(execution_arn: str) -> Response:
         """Creates a ResourceNotFound response for durable executions"""
         exception_tuple = LambdaErrorResponses.ResourceNotFoundException
         return BaseLocalService.service_response(
@@ -290,7 +293,7 @@ def durable_execution_not_found(execution_arn):
         )
 
     @staticmethod
-    def _construct_headers(error_type):
+    def _construct_headers(error_type: str) -> Dict[str, str]:
         """
         Constructs Headers for the Local Lambda Error Response