feat: add support for ResponseTransferMode for API gateway events

Author: bchampp

samtranslator/internal/schema_source/aws_serverless_function.py

@@ -287,6 +287,7 @@ class ApiEventProperties(BaseModel):
     RequestParameters: Optional[RequestModelProperty] = apieventproperties("RequestParameters")
     RestApiId: Optional[Union[str, Ref]] = apieventproperties("RestApiId")
     TimeoutInMillis: Optional[PassThroughProp]  # TODO: add doc
+    ResponseTransferMode: Optional[PassThroughProp] = apieventproperties("ResponseTransferMode")
 
 
 class ApiEvent(BaseModel):

samtranslator/internal/schema_source/sam-docs.json

@@ -155,7 +155,8 @@
       "RequestModel": "Request model to use for this specific Api\\+Path\\+Method\\. This should reference the name of a model specified in the `Models` section of an [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource\\.  \n*Type*: [RequestModel](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-function-requestmodel.html)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
       "RequestParameters": "Request parameters configuration for this specific Api\\+Path\\+Method\\. All parameter names must start with `method.request` and must be limited to `method.request.header`, `method.request.querystring`, or `method.request.path`\\.  \nA list can contain both parameter name strings and [RequestParameter](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-function-requestparameter.html) objects\\. For strings, the `Required` and `Caching` properties will default to `false`\\.  \n*Type*: List of \\[ String \\| [RequestParameter](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-function-requestparameter.html) \\]  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
       "RestApiId": "Identifier of a RestApi resource, which must contain an operation with the given path and method\\. Typically, this is set to reference an [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource defined in this template\\.  \nIf you don't define this property, AWS SAM creates a default [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource using a generated `OpenApi` document\\. That resource contains a union of all paths and methods defined by `Api` events in the same template that do not specify a `RestApiId`\\.  \nThis cannot reference an [AWS::Serverless::Api](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html) resource defined in another template\\.  \n*Type*: String  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
-      "TimeoutInMillis": "Custom timeout between 50 and 29,000 milliseconds\\.  \nWhen you specify this property, AWS SAM modifies your OpenAPI definition\\. The OpenAPI definition must be specified inline using the `DefinitionBody` property\\. \n*Type*: Integer  \n*Required*: No  \n*Default*: 29,000 milliseconds or 29 seconds  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\."
+      "TimeoutInMillis": "Custom timeout between 50 and 29,000 milliseconds\\.  \nWhen you specify this property, AWS SAM modifies your OpenAPI definition\\. The OpenAPI definition must be specified inline using the `DefinitionBody` property\\. \n*Type*: Integer  \n*Required*: No  \n*Default*: 29,000 milliseconds or 29 seconds  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
+      "ResponseTransferMode": "The response transfer mode for the Lambda function integration. Set to `RESPONSE_STREAM` to enable Lambda response streaming through API Gateway, allowing the function to stream responses back to clients. When set to `RESPONSE_STREAM`, API Gateway uses the Lambda InvokeWithResponseStreaming API.  \n*Type*: String  \n*Required*: No  \n*Valid values*: `BUFFERED` \\| `RESPONSE_STREAM`  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`responseTransferMode`](https://docs.aws.amazon.com/apigateway/latest/api/API_Integration.html#responseTransferMode) property of an `AWS::ApiGateway::Method Integration`\\."
     },
     "sam-property-function-apifunctionauth": {
       "ApiKeyRequired": "Requires an API key for this API, path, and method\\.  \n*Type*: Boolean  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\.",
@@ -422,7 +423,7 @@
       "LoggingConfig": "A configuration object that specifies the logging configuration for the event source mapping\\.  \n*Type*: [LoggingConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-loggingconfig.html)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`LoggingConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-loggingconfig.html) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
       "MetricsConfig": "A configuration object that specifies the metrics configuration for the event source mapping\\.  \n*Type*: [MetricsConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-metricsconfig.html)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`MetricsConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-metricsconfig.html) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
       "ProvisionedPollerConfig": "A configuration object that specifies the provisioned poller configuration for the event source mapping\\.  \n*Type*: [ProvisionedPollerConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-provisionedpollerconfig.html)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`ProvisionedPollerConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-provisionedpollerconfig.html) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
-      "SchemaRegistryConfig": "A configuration object that specifies the schema registry configuration for the event source mapping\\.  \n*Type*: [SchemaRegistryConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-schemaregistryconfig.html)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`SchemaRegistryConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-schemaregistryconfig.html) property of an `AWS::Lambda::EventSourceMapping` resource\\.",      
+      "SchemaRegistryConfig": "A configuration object that specifies the schema registry configuration for the event source mapping\\.  \n*Type*: [SchemaRegistryConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-schemaregistryconfig.html)  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`SchemaRegistryConfig`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-schemaregistryconfig.html) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
       "BisectBatchOnFunctionError": "If the function returns an error, split the batch in two and retry\\.  \n*Type*: Boolean  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`BisectBatchOnFunctionError`](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-bisectbatchonfunctionerror) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
       "FunctionResponseTypes": "A list of the response types currently applied to the event source mapping\\. For more information, see [Reporting batch item failures](https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html#services-ddb-batchfailurereporting) in the *AWS Lambda Developer Guide*\\.  \n*Valid values*: `ReportBatchItemFailures`  \n*Type*: List  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`FunctionResponseTypes`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-functionresponsetypes) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
       "MaximumRecordAgeInSeconds": "The maximum age of a record that Lambda sends to a function for processing\\.  \n*Type*: Integer  \n*Required*: No  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`MaximumRecordAgeInSeconds`](http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-maximumrecordageinseconds) property of an `AWS::Lambda::EventSourceMapping` resource\\.",
@@ -853,4 +854,4 @@
       "UseAliasAsEventTarget": "Indicate whether or not to pass the alias, created by using the `AutoPublishAlias` property, to the events source's target defined with [Events](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/#sam-statemachine-events.html#sam-statemachine-events)\\.  \nSpecify `True` to use the alias as the events' target\\.  \n*Type*: Boolean  \n*Required*: No  \n*Default*: `False`  \n*AWS CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an AWS CloudFormation equivalent\\."
     }
   }
-}
+}

samtranslator/model/eventsources/push.py

@@ -689,6 +689,7 @@ class Api(PushEventSource):
         "RequestModel": PropertyType(False, IS_DICT),
         "RequestParameters": PropertyType(False, IS_LIST),
         "TimeoutInMillis": PropertyType(False, IS_INT),
+        "ResponseTransferMode": PropertyType(False, IS_STR),
     }
 
     Path: str
@@ -699,6 +700,7 @@ class Api(PushEventSource):
     RequestModel: Optional[Dict[str, Any]]
     RequestParameters: Optional[List[Any]]
     TimeoutInMillis: Optional[PassThrough]
+    ResponseTransferMode: Optional[PassThrough]
 
     def resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]:
         """
@@ -863,7 +865,7 @@ def _add_swagger_integration(  # type: ignore[no-untyped-def] # noqa: PLR0912, P
             swagger_body = SwaggerEditor.gen_skeleton()
 
         partition = ArnGenerator.get_partition_name()
-        uri = _build_apigw_integration_uri(function, partition)  # type: ignore[no-untyped-call]
+        uri = _build_apigw_integration_uri(function, partition, self.ResponseTransferMode)  # type: ignore[no-untyped-call]
 
         editor = SwaggerEditor(swagger_body)
 
@@ -882,7 +884,15 @@ def _add_swagger_integration(  # type: ignore[no-untyped-def] # noqa: PLR0912, P
         sam_expect(method_auth, self.relative_id, "Auth", is_sam_event=True).to_be_a_map()
         api_auth = api.get("Auth") or Py27Dict()
         sam_expect(api_auth, api_id, "Auth").to_be_a_map()
-        editor.add_lambda_integration(self.Path, self.Method, uri, method_auth, api_auth, condition=condition)
+        editor.add_lambda_integration(
+            self.Path,
+            self.Method,
+            uri,
+            method_auth,
+            api_auth,
+            condition=condition,
+            invoke_mode=self.ResponseTransferMode,
+        )
 
         # self.Stage is not None as it is set in _get_permissions()
         # before calling this method.
@@ -1550,14 +1560,23 @@ def _add_auth_to_openapi_integration(
         editor.add_auth_to_method(api=api, path=self._path, method_name=self._method, auth=self.Auth)  # type: ignore[no-untyped-call]
 
 
-def _build_apigw_integration_uri(function, partition):  # type: ignore[no-untyped-def]
+def _build_apigw_integration_uri(function, partition, response_transfer_mode=None):  # type: ignore[no-untyped-def]
     function_arn = function.get_runtime_attr("arn")
+    # Use response-streaming-invocations path when ResponseTransferMode is RESPONSE_STREAM
+    # See: https://aws.amazon.com/blogs/compute/building-responsive-apis-with-amazon-api-gateway-response-streaming/
+    if response_transfer_mode == "RESPONSE_STREAM":
+        api_version = "2021-11-15"
+        invocation_path = "/response-streaming-invocations"
+    else:
+        api_version = "2015-03-31"
+        invocation_path = "/invocations"
+
     arn = (
         "arn:"
         + partition
-        + ":apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/"
+        + f":apigateway:${{AWS::Region}}:lambda:path/{api_version}/functions/"
         + make_shorthand(function_arn)
-        + "/invocations"
+        + invocation_path
     )
     # function_arn is always of the form {"Fn::GetAtt": ["<function_logical_id>", "Arn"]}.
     # We only want to check if the function logical id is a Py27UniStr instance.

samtranslator/open_api/open_api.py

@@ -114,7 +114,14 @@ def is_integration_function_logical_id_match(self, path_name, method_name, logic
         return True
 
     def add_lambda_integration(  # type: ignore[no-untyped-def] # noqa: PLR0913
-        self, path, method, integration_uri, method_auth_config=None, api_auth_config=None, condition=None
+        self,
+        path,
+        method,
+        integration_uri,
+        method_auth_config=None,
+        api_auth_config=None,
+        condition=None,
+        invoke_mode=None,
     ):
         """
         Adds aws_proxy APIGW integration to the given path+method.
@@ -147,6 +154,9 @@ def add_lambda_integration(  # type: ignore[no-untyped-def] # noqa: PLR0913
             path_item[method][self._X_APIGW_INTEGRATION]["payloadFormatVersion"] = "2.0"
             path_item[method][self._X_APIGW_INTEGRATION]["uri"] = integration_uri
 
+            if invoke_mode:
+                path_item[method][self._X_APIGW_INTEGRATION]["invokeMode"] = invoke_mode
+
             if path == self._DEFAULT_PATH and method == self._X_ANY_METHOD:
                 path_item[method]["isDefaultRoute"] = True
 

samtranslator/schema/schema.json

@@ -356608,6 +356608,15 @@
           "title": "RequestParameters",
           "type": "array"
         },
+        "ResponseTransferMode": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/PassThroughProp"
+            }
+          ],
+          "markdownDescription": "The response transfer mode for the Lambda function integration. Set to `RESPONSE_STREAM` to enable Lambda response streaming through API Gateway, allowing the function to stream responses back to clients. When set to `RESPONSE_STREAM`, API Gateway uses the Lambda InvokeWithResponseStreaming API.  \n*Type*: String  \n*Required*: No  \n*Valid values*: `BUFFERED` \\| `RESPONSE_STREAM`  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`responseTransferMode`](https://docs.aws.amazon.com/apigateway/latest/api/API_Integration.html#responseTransferMode) property of an `AWS::ApiGateway::Method Integration`\\.",
+          "title": "ResponseTransferMode"
+        },
         "RestApiId": {
           "anyOf": [
             {

samtranslator/swagger/swagger.py

@@ -125,6 +125,7 @@ def add_lambda_integration(  # noqa: PLR0913
         method_auth_config: Dict[str, Any],
         api_auth_config: Dict[str, Any],
         condition: Optional[str] = None,
+        invoke_mode: Optional[Any] = None,
     ) -> None:
         """
         Adds aws_proxy APIGW integration to the given path+method.
@@ -150,6 +151,10 @@ def add_lambda_integration(  # noqa: PLR0913
             path_item[method][self._X_APIGW_INTEGRATION]["httpMethod"] = "POST"
             path_item[method][self._X_APIGW_INTEGRATION]["uri"] = _integration_uri
 
+            # When using RESPONSE_STREAM invoke mode, set responseTransferMode to STREAM
+            if invoke_mode == "RESPONSE_STREAM":
+                path_item[method][self._X_APIGW_INTEGRATION]["responseTransferMode"] = "STREAM"
+
             if (
                 method_auth_config.get("Authorizer") == "AWS_IAM"
                 or api_auth_config.get("DefaultAuthorizer") == "AWS_IAM"

schema_source/sam.schema.json

@@ -5638,6 +5638,15 @@
           "title": "RequestParameters",
           "type": "array"
         },
+        "ResponseTransferMode": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/PassThroughProp"
+            }
+          ],
+          "markdownDescription": "The response transfer mode for the Lambda function integration. Set to `RESPONSE_STREAM` to enable Lambda response streaming through API Gateway, allowing the function to stream responses back to clients. When set to `RESPONSE_STREAM`, API Gateway uses the Lambda InvokeWithResponseStreaming API.  \n*Type*: String  \n*Required*: No  \n*Valid values*: `BUFFERED` \\| `RESPONSE_STREAM`  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`responseTransferMode`](https://docs.aws.amazon.com/apigateway/latest/api/API_Integration.html#responseTransferMode) property of an `AWS::ApiGateway::Method Integration`\\.",
+          "title": "ResponseTransferMode"
+        },
         "RestApiId": {
           "anyOf": [
             {

tests/translator/input/api_with_invoke_mode.yaml

@@ -0,0 +1,20 @@
+Resources:
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/stream.zip
+      Handler: index.handler
+      Runtime: nodejs20.x
+      Events:
+        StreamApi:
+          Type: Api
+          Properties:
+            Path: /stream
+            Method: post
+            ResponseTransferMode: RESPONSE_STREAM
+        BufferedApi:
+          Type: Api
+          Properties:
+            Path: /buffered
+            Method: get
+            ResponseTransferMode: BUFFERED