New serverless pattern - apigw dynamodb python cdk

apigw-dynamodb-python-cdk/README.md

@@ -0,0 +1,82 @@
+
+# API Gateway direct integration to DynamoDB
+
+This pattern shows how to create an API Gateway with direct integration to DynamoDB. 
+The pettern showcase transformation of request/response using VTL and CDK and implement examples for using Cognito, Lambda authorizer and API keys. 
+
+Learn more about this pattern at Serverless Land Patterns: [Serverless Land Patterns](https://serverlessland.com/patterns/apigw-dynamodb-python-cdk).
+
+Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. You are responsible for any AWS costs incurred. No warranty is implied in this example.
+
+![alt text](image.png)
+
+## Requirements
+
+* [Create an AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) if you do not already have one and log in. The IAM user that you use must have sufficient permissions to make necessary AWS service calls and manage AWS resources.
+* [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git)
+* [AWS Cloud Development Kit](https://docs.aws.amazon.com/cdk/latest/guide/cli.html) (AWS CDK) installed
+
+## Deployment Instructions
+
+1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository:
+```
+git clone https://github.com/aws-samples/serverless-patterns/ 
+```
+2. Change directory
+```
+cd serverless-patterns/apigw-dynamodb-python-cdk
+```
+3. To manually create a virtualenv on MacOS and Linux:
+```
+python3 -m venv .venv
+```
+4. After the init process completes and the virtualenv is created, you can use the following to activate virtualenv.
+```
+source .venv/bin/activate
+``` 
+6. After activating your virtual environment for the first time, install the app's standard dependencies:
+```
+python -m pip install -r requirements.txt
+```
+7. Install jwt package for Lambda:
+```
+cd src; pip install pyjwt --target . 
+```
+8. Zip the Lambda function and dependencies 
+```
+zip -r lambda.zip .
+```
+9. To generate a cloudformation templates (optional)
+```
+cdk synth
+```
+10. To deploy AWS resources as a CDK project
+```
+cdk deploy 
+```
+
+## How it works
+At the end of the deployment the CDK output will list stack outputs, and an API Gateway URL. In the customer's AWS account, a REST API along with an authorizer, Cognito user pool, and a DynamoDB table will be created.
+Put resource - uses Lambda authorizer to authenticate the client and send allow/deny to API Gateway. 
+Get resource - uses API key to control the rate limit. Need to provide valid key for the request with x-api-key header.
+Delete resource - uses Cognito to authenticate the client. Cognito token need to be provided with Authorization header. 
+
+## Testing
+1. Run pytest
+```
+pytest tests/test_apigw_dynamodb_python_stack.py 
+```
+## Cleanup
+
+1. Delete the stack
+    ```bash
+   cdk destroy
+    ```
+1. Confirm the stack has been deleted
+    ```bash
+    cdk list
+    ```
+----
+Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+SPDX-License-Identifier: MIT-0

apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/api_key_usage_plan_construct.py

@@ -0,0 +1,62 @@
+from constructs import Construct
+import aws_cdk.aws_apigateway as apigateway
+
+
+class UsagePlanConstruct(Construct):
+    def __init__(self, scope: Construct, id: str, apigateway_construct, plan_name, plan_config ,**kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        # Map the period of the usage plan from the config to apigateway.Period.XXX
+        period_enum = self.get_period_enum(plan_config['quota']['period'])
+
+        # Create usage plan dynamically using the context data
+        usage_plan = apigateway_construct.api.add_usage_plan(plan_name,
+            name=plan_name,
+            throttle=apigateway.ThrottleSettings(
+                rate_limit=plan_config['throttle']['rate_limit'],
+                burst_limit=plan_config['throttle']['burst_limit']
+            ),
+            quota=apigateway.QuotaSettings(
+                limit=plan_config['quota']['limit'],
+                period=period_enum
+            )
+        )
+
+        # Create API key 
+        api_key = apigateway.ApiKey(self, f"ApiKey-{plan_name}",
+                                    api_key_name=f"ApiKey-{plan_name}")
+        self.api_key_id = api_key.key_id
+        usage_plan.add_api_key(api_key)
+
+        # If method is configured in the context assign the API key to the relevant API method
+        if plan_config['method']:
+            def get_method(method_name):
+                method_mapping = { # Change the method to fit your API
+                    "GET": apigateway_construct.get_method,
+                    "POST": apigateway_construct.put_method,
+                    "DELETE": apigateway_construct.delete_method
+                }
+                return method_mapping.get(method_name.upper())
+            usage_plan.add_api_stage(
+            stage=apigateway_construct.api.deployment_stage,
+            throttle=[apigateway.ThrottlingPerMethod(
+                method=get_method(plan_config['method']),
+                throttle=apigateway.ThrottleSettings(
+                    rate_limit=100,
+                    burst_limit=1
+                ))]
+            )
+
+
+
+
+    @staticmethod
+    def get_period_enum(period: str) -> apigateway.Period:
+        period_mapping = {
+            "DAY": apigateway.Period.DAY,
+            "WEEK": apigateway.Period.WEEK,
+            "MONTH": apigateway.Period.MONTH
+        }
+        return period_mapping.get(period.upper())
+
+

apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/apigateway_construct.py

@@ -0,0 +1,204 @@
+from constructs import Construct
+import aws_cdk.aws_apigateway as apigateway
+import aws_cdk.aws_iam as iam
+import os
+
+class ApiGatewayConstruct(Construct):
+    def __init__(self, scope: Construct, id: str, cognito_construct, dynamodb_construct, lambda_construct, vtl_dir ,**kwargs) -> None:
+        super().__init__(scope, id, **kwargs)
+
+        self.vtl_dir = vtl_dir
+
+        # Define the Cognito Authorizer
+        cognito_authorizer = apigateway.CognitoUserPoolsAuthorizer(self, "CognitoAuthorizer",
+            cognito_user_pools=[cognito_construct.user_pool]
+        )
+
+        # Define lambda authorizer
+        lambda_authorizer = apigateway.RequestAuthorizer(self, "LambdaAuthorizer",
+            handler=lambda_construct.lambda_function,
+            identity_sources=[apigateway.IdentitySource.header("Authorization")]
+        )
+
+        # Create IAM role
+        api_gateway_role = iam.Role(self, "ApiGatewayDynamoDBRole",
+            assumed_by=iam.ServicePrincipal("apigateway.amazonaws.com"),
+            inline_policies={
+                "DynamoDBAccess": iam.PolicyDocument(
+                    statements=[
+                        iam.PolicyStatement(
+                            actions=["dynamodb:PutItem","dynamodb:DeleteItem", "dynamodb:Scan", "dynamodb:Query", "dynamodb:DescribeTable"],
+                            resources=[dynamodb_construct.table.table_arn]
+                        )
+                    ]
+                )
+            }
+        )
+
+        # Define API Gateway
+        self.api = apigateway.RestApi(self, "MyApi",
+            rest_api_name="My Service",
+            description="This service serves my DynamoDB table.",
+            cloud_watch_role=True,
+            deploy_options=apigateway.StageOptions(
+                stage_name="prod",
+                logging_level=apigateway.MethodLoggingLevel.INFO,
+                data_trace_enabled=True,
+                metrics_enabled=True,
+                variables={
+                    "TableName": dynamodb_construct.table.table_name}
+            )
+        )
+
+        # Change default response for Bad Request Body
+        self.api.add_gateway_response(
+            "BadRequestBody",
+            type=apigateway.ResponseType.BAD_REQUEST_BODY,
+            templates={
+                "application/json": '{"message": "Invalid Request Body: $context.error.validationErrorString"}'
+            }
+        )
+
+        # Create request model schema
+        request_model_schema = apigateway.JsonSchema(
+            type=apigateway.JsonSchemaType.OBJECT,
+            required=["ID","FirstName", "Age"],
+            properties={
+                "ID": {"type": apigateway.JsonSchemaType.STRING},
+                "FirstName": {"type": apigateway.JsonSchemaType.STRING},
+                "Age": {"type": apigateway.JsonSchemaType.NUMBER}
+            },
+            # Allow to send additional properites - handled in putItem.vtl to construct them to the request
+            additional_properties=True
+        )
+
+        # Create a request validator
+        request_validator = apigateway.RequestValidator(self, "RequestValidator",
+            rest_api=self.api,
+            validate_request_body=True,
+            validate_request_parameters=False
+        )
+
+        # Create the request model
+        request_model = apigateway.Model(self, "RequestModel",
+            rest_api=self.api,
+            content_type="application/json",
+            schema=request_model_schema,
+            model_name="PutObjectRequestModel"
+        )
+
+        # Create integration request
+        integration_request = apigateway.AwsIntegration(
+            service="dynamodb",
+            action="PutItem",
+            options=apigateway.IntegrationOptions(
+                credentials_role=api_gateway_role,
+                request_templates={
+                    "application/json": 
+                    self.get_vtl_template("putItem.vtl")
+                },
+                integration_responses=[
+                    apigateway.IntegrationResponse(
+                        status_code="200",
+                        response_templates={
+                            "application/json": self.get_vtl_template("response.vtl")
+                        }
+                    ),
+                ]
+            )
+        )
+
+        # Create a resource and method for the API Gateway
+        put_resource = self.api.root.add_resource("put")
+        self.put_method = put_resource.add_method(
+            "POST",
+            integration_request,
+            authorization_type=apigateway.AuthorizationType.CUSTOM,
+            authorizer=lambda_authorizer,
+            request_validator=request_validator,
+            request_models={"application/json": request_model},
+            method_responses=[
+                apigateway.MethodResponse(status_code="200",response_models={
+                        "application/json": apigateway.Model.EMPTY_MODEL
+                    }   ),
+            ]
+        )
+
+        # Add GET method with response mapping
+        get_integration = apigateway.AwsIntegration(
+            service="dynamodb",
+            action="Scan",
+            options=apigateway.IntegrationOptions(
+                credentials_role=api_gateway_role,
+                request_templates={
+                    "application/json": self.get_vtl_template('scan_request.vtl')
+                },
+                integration_responses=[
+                    apigateway.IntegrationResponse(
+                        status_code="200",
+                        response_templates={
+                            "application/json": self.get_vtl_template('scan.vtl')
+                        }
+                    ),
+                ]
+            )
+        )
+
+        get_resource = self.api.root.add_resource('get')
+        self.get_method = get_resource.add_method(
+            "GET", get_integration,
+            api_key_required=True,
+            method_responses=[
+                apigateway.MethodResponse(
+                    status_code="200",
+                    response_models={
+                        "application/json": apigateway.Model.EMPTY_MODEL
+                    }
+                ),
+            ]
+        )
+
+        delete_resource = self.api.root.add_resource('delete')
+        delete_resource_id = delete_resource.add_resource('{id}')
+        self.delete_method = delete_resource_id.add_method(
+            "POST",
+            apigateway.AwsIntegration(
+                service="dynamodb",
+                action="DeleteItem",
+                options=apigateway.IntegrationOptions(
+                    credentials_role=api_gateway_role,
+                    request_templates={
+                        "application/json": 
+                        self.get_vtl_template("deleteItem.vtl")
+                    },
+                    integration_responses=[
+                        apigateway.IntegrationResponse(
+                            status_code="200",
+                            response_templates={
+                                "application/json": '{"message": "Item deleted"}'
+                            }
+                        ),
+                    ]
+                )
+            ),
+            authorization_type=apigateway.AuthorizationType.COGNITO,
+            authorizer=cognito_authorizer,
+            request_validator=request_validator,
+            method_responses=[
+                apigateway.MethodResponse(
+                    status_code="200",
+                    response_models={
+                        "application/json": apigateway.Model.EMPTY_MODEL
+                    }
+                ),
+            ]
+        )
+
+
+    def get_vtl_template(self, filename: str) -> str:
+        """
+        Reads a VTL template from a file and returns its contents as a string.
+        """
+        template_path = os.path.join(self.vtl_dir, filename)
+        with open(template_path, "r") as f:
+            return f.read()