diff --git a/apigw-dynamodb-python-cdk/README.md b/apigw-dynamodb-python-cdk/README.md
new file mode 100644
index 000000000..71cffe310
--- /dev/null
+++ b/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 pattern 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.
+
+
+
+## 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 . ; cd
+```
+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
diff --git a/apigw-dynamodb-python-cdk/apigw-dynamodb-python-cdk.json b/apigw-dynamodb-python-cdk/apigw-dynamodb-python-cdk.json
new file mode 100644
index 000000000..26f79f7e2
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/apigw-dynamodb-python-cdk.json
@@ -0,0 +1,71 @@
+{
+ "title": "API Gateway direct integration to DynamoDB",
+ "description": "Direct integration with API Gateway to DynamoDB with transformation using VTL and CDK and examples for Cognito, Lambda authorizer and API keys.",
+ "language": "Python",
+ "level": "300",
+ "framework": "CDK",
+ "introBox": {
+ "headline": "How it works",
+ "text": [
+ "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."
+ ]
+ },
+ "gitHub": {
+ "template": {
+ "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/apigw-dynamodb-python-cdk",
+ "templateURL": "serverless-patterns/apigw-dynamodb-python-cdk",
+ "projectFolder": "apigw-dynamodb-python-cdk",
+ "templateFile": "apigw_dynamodb_python_cdk_stack.py"
+ }
+ },
+ "resources": {
+ "bullets": [
+ {
+ "text": "API Gateway Integrations",
+ "link": "https://docs.aws.amazon.com/apigateway/latest/developerguide/how-to-integration-settings.html"
+ }
+ ]
+ },
+ "deploy": {
+ "text": [
+ "cdk deploy"
+ ]
+ },
+ "testing": {
+ "text": [
+ "See the GitHub repo for detailed testing instructions."
+ ]
+ },
+ "cleanup": {
+ "text": [
+ "Delete the stack: cdk delete."
+ ]
+ },
+ "authors": [
+ {
+ "name": "Maya Morav Freiman",
+ "image": "https://avatars.githubusercontent.com/u/11615439?v=4",
+ "bio": "Technical Account Manager at AWS",
+ "linkedin": "mayaaws"
+ }
+ ],
+ "patternArch": {
+ "icon1": {
+ "x": 20,
+ "y": 50,
+ "service": "apigw",
+ "label": "API Gateway REST API"
+ },
+ "icon2": {
+ "x": 80,
+ "y": 50,
+ "service": "dynamodb",
+ "label": "Amazon DynamoDB"
+ },
+ "line1": {
+ "from": "icon1",
+ "to": "icon2"
+ }
+ }
+}
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/__init__.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/api_key_usage_plan_construct.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/api_key_usage_plan_construct.py
new file mode 100644
index 000000000..d3ad2ab6f
--- /dev/null
+++ b/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())
+
+
\ No newline at end of file
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/apigateway_construct.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/apigateway_construct.py
new file mode 100644
index 000000000..c2f71cd0e
--- /dev/null
+++ b/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()
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/apigw_dynamodb_python_cdk_stack.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/apigw_dynamodb_python_cdk_stack.py
new file mode 100644
index 000000000..69fd41c76
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/apigw_dynamodb_python_cdk_stack.py
@@ -0,0 +1,58 @@
+from aws_cdk import Stack
+from aws_cdk import CfnOutput
+from constructs import Construct
+
+from apigw_dynamodb_python_cdk.api_key_usage_plan_construct import UsagePlanConstruct
+from apigw_dynamodb_python_cdk.apigateway_construct import ApiGatewayConstruct
+from apigw_dynamodb_python_cdk.cognito_construct import CognitoConstruct
+from apigw_dynamodb_python_cdk.dynamodb_construct import DynamoDBConstruct
+from apigw_dynamodb_python_cdk.lambda_construct import LambdaConstruct
+from apigw_dynamodb_python_cdk.user_pool_group_construct import UserPoolGroupConstruct
+
+class ApigwDynamodbPythonStack(Stack):
+
+ def __init__(self, scope: Construct, id: str, **kwargs) -> None:
+ super().__init__(scope, id, **kwargs)
+
+ vtl_dir = self.node.try_get_context("vtl_dir")
+
+
+ lambda_construct = LambdaConstruct(self, "LambdaConstruct")
+ cognito_construct = CognitoConstruct(self, "CognitoConstruct")
+ dynamodb_construct = DynamoDBConstruct(self, "DynamoDBConstruct")
+ # Passing full construct is an option, specific ID can be used - Example: dynamodb_construct.table.table_arn
+ apigateway_construct = ApiGatewayConstruct(self, "ApiGatewayConstruct", cognito_construct, dynamodb_construct, lambda_construct, vtl_dir)
+
+ # Using the context defined in app.py to iterate and create multiple resources
+ group_names = self.node.try_get_context("group_names")
+ if group_names:
+ for group_name in group_names:
+ UserPoolGroupConstruct(
+ self,
+ f"UserPoolGroup{group_name}Construct",
+ cognito_construct,
+ group_name
+ )
+
+
+ api_key_ids = []
+ usage_plans = self.node.try_get_context("usage_plans")
+ if usage_plans:
+ for usage_plan_name, usage_plan_config in usage_plans.items():
+ use_plan_construct = UsagePlanConstruct(
+ self,
+ f"ApiGateway{usage_plan_name}Construct",
+ apigateway_construct,
+ usage_plan_name,
+ usage_plan_config
+ )
+ api_key_ids.append(use_plan_construct.api_key_id)
+
+ for index, api_key_id in enumerate(api_key_ids):
+ CfnOutput(self, f"ApiKeyId{index}", value=api_key_id)
+ # Outputs - used also by the tests
+ CfnOutput(self, "CognitoUserPoolId", value=cognito_construct.user_pool.user_pool_id)
+ CfnOutput(self, "CognitoClientId", value=cognito_construct.user_pool_client.user_pool_client_id)
+ CfnOutput(self, "ApiUrl", value=apigateway_construct.api.url)
+ CfnOutput(self, "DynamoDBTableName", value=dynamodb_construct.table.table_name)
+
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/cognito_construct.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/cognito_construct.py
new file mode 100644
index 000000000..97d8bb2cd
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/cognito_construct.py
@@ -0,0 +1,54 @@
+import aws_cdk as cdk
+from constructs import Construct
+import aws_cdk.aws_cognito as cognito
+
+
+class CognitoConstruct(Construct):
+ def __init__(self, scope: Construct, id: str, **kwargs) -> None:
+ super().__init__(scope, id, **kwargs)
+
+ # Create Cognito user pool
+ self.user_pool = cognito.UserPool(self, "MyUserPool",
+ user_pool_name="my_user_pool",
+ self_sign_up_enabled=True,
+ auto_verify=cognito.AutoVerifiedAttrs(email=True),
+ sign_in_aliases=cognito.SignInAliases(email=True),
+ standard_attributes={
+ "email": {
+ "required": True,
+ "mutable": False
+ }
+ },
+ removal_policy=cdk.RemovalPolicy.DESTROY
+ )
+
+ # Create user pool client
+ self.user_pool_client = cognito.UserPoolClient(self, "UserPoolClient",
+ user_pool=self.user_pool,
+ generate_secret=False,
+ auth_flows=cognito.AuthFlow(
+ user_password=True,
+ admin_user_password=True,
+ # user_srp=True,
+ ),
+ o_auth=cognito.OAuthSettings(
+ callback_urls=["http://localhost"],
+ flows=cognito.OAuthFlows(
+ authorization_code_grant=True
+ ),
+ scopes=[
+ cognito.OAuthScope.EMAIL,
+ cognito.OAuthScope.OPENID,
+ cognito.OAuthScope.COGNITO_ADMIN
+ ]
+ ),
+ supported_identity_providers=[cognito.UserPoolClientIdentityProvider.COGNITO]
+ )
+
+ # Define the user pool domain
+ cognito.UserPoolDomain(self, "UserPoolDomain_",
+ user_pool=self.user_pool,
+ cognito_domain=cognito.CognitoDomainOptions(
+ domain_prefix="a1faegn" # This must be unique across all AWS accounts and regions
+ )
+ )
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/dynamodb_construct.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/dynamodb_construct.py
new file mode 100644
index 000000000..c49dc55e4
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/dynamodb_construct.py
@@ -0,0 +1,17 @@
+import aws_cdk as cdk
+from constructs import Construct
+import aws_cdk.aws_dynamodb as dynamodb
+
+
+class DynamoDBConstruct(Construct):
+ def __init__(self, scope: Construct, id: str, **kwargs) -> None:
+ super().__init__(scope, id, **kwargs)
+
+ # Create DynamoDB table
+ self.table = dynamodb.Table(
+ self, "MyTable",
+ partition_key=dynamodb.Attribute(name="ID", type=dynamodb.AttributeType.STRING),
+ sort_key=dynamodb.Attribute(name="FirstName", type=dynamodb.AttributeType.STRING),
+ removal_policy=cdk.RemovalPolicy.DESTROY, # NOT recommended for production code
+ billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST
+ )
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/lambda_construct.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/lambda_construct.py
new file mode 100644
index 000000000..863b8761b
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/lambda_construct.py
@@ -0,0 +1,38 @@
+from constructs import Construct
+import aws_cdk.aws_lambda as lambda_
+import aws_cdk.aws_iam as iam
+
+class LambdaConstruct(Construct):
+ def __init__(self, scope: Construct, id: str ,**kwargs) -> None:
+ super().__init__(scope, id, **kwargs)
+
+ # Create lambda execution role
+ lambda_execution_role = iam.Role(
+ self,
+ "LambdaExecutionRole",
+ assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
+ managed_policies=[
+ iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole")
+ ],
+ inline_policies={
+ "LambdaPolicy": iam.PolicyDocument(
+ statements=[
+ iam.PolicyStatement(
+ effect=iam.Effect.ALLOW,
+ actions=["apigateway:GET"],
+ resources=["*"]
+ )
+ ]
+ )
+ }
+ )
+
+ # Create lambda function.
+ self.lambda_function = lambda_.Function(self, "LambdaFunction",
+ runtime=lambda_.Runtime.PYTHON_3_12,
+ handler="lambda_function.lambda_handler",
+ role=lambda_execution_role,
+ code=lambda_.Code.from_asset("src/lambda.zip")
+ )
+
+
\ No newline at end of file
diff --git a/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/user_pool_group_construct.py b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/user_pool_group_construct.py
new file mode 100644
index 000000000..32ea8b53d
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/apigw_dynamodb_python_cdk/user_pool_group_construct.py
@@ -0,0 +1,18 @@
+from constructs import Construct
+import aws_cdk.aws_cognito as cognito
+
+
+class UserPoolGroupConstruct(Construct):
+ def __init__(self, scope: Construct, id: str, cognito_construct, group_name, **kwargs) -> None:
+ super().__init__(scope, id, **kwargs)
+ # Create user pool group.
+ # Required parameters -
+ # 1. User pool ID - taken from the cofnito construct
+ # 2. Group name - taken from the stack context
+ cognito.CfnUserPoolGroup(self, group_name,
+ user_pool_id=cognito_construct.user_pool.user_pool_id,
+ group_name=group_name,
+ description=f"Group created {group_name}",
+ precedence=1
+ )
+
diff --git a/apigw-dynamodb-python-cdk/app.py b/apigw-dynamodb-python-cdk/app.py
new file mode 100644
index 000000000..1e5e23fbf
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/app.py
@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+import os
+import aws_cdk as cdk
+from apigw_dynamodb_python_cdk.apigw_dynamodb_python_cdk_stack import ApigwDynamodbPythonStack
+
+
+app = cdk.App()
+
+vtl_dir = os.path.join(os.path.dirname(__file__), "vtl")
+group_names = ["Group-FreeTier", "Group-BasicUsagePlan"]
+
+app.node.set_context("group_names", group_names)
+app.node.set_context("vtl_dir", vtl_dir)
+
+usage_plans = {
+ "FreeTier": {
+ "quota": {
+ "limit": 500,
+ "period": "DAY"
+ },
+ "throttle": {
+ "burst_limit": 10,
+ "rate_limit": 5
+ },
+ "method": "GET"
+ },
+ "BasicUsagePlan": {
+ "quota": {
+ "limit": 10000,
+ "period": "MONTH"
+ },
+ "throttle": {
+ "burst_limit": 100,
+ "rate_limit": 50
+ },
+ "method": "POST"
+ }
+}
+
+app.node.set_context("usage_plans", usage_plans)
+
+stack = ApigwDynamodbPythonStack(app, "ApigwDynamodbPythonStack")
+
+app.synth()
diff --git a/apigw-dynamodb-python-cdk/cdk.json b/apigw-dynamodb-python-cdk/cdk.json
new file mode 100644
index 000000000..0e9b4305b
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/cdk.json
@@ -0,0 +1,64 @@
+{
+ "app": "python3 app.py",
+ "watch": {
+ "include": [
+ "**"
+ ],
+ "exclude": [
+ "README.md",
+ "cdk*.json",
+ "requirements*.txt",
+ "source.bat",
+ "**/__init__.py",
+ "**/__pycache__",
+ "tests"
+ ]
+ },
+ "context": {
+ "@aws-cdk/aws-lambda:recognizeLayerVersion": true,
+ "@aws-cdk/core:checkSecretUsage": true,
+ "@aws-cdk/core:target-partitions": [
+ "aws",
+ "aws-cn"
+ ],
+ "@aws-cdk-containers/ecs-service-extensions:enableDefaultLogDriver": true,
+ "@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true,
+ "@aws-cdk/aws-ecs:arnFormatIncludesClusterName": true,
+ "@aws-cdk/aws-iam:minimizePolicies": true,
+ "@aws-cdk/core:validateSnapshotRemovalPolicy": true,
+ "@aws-cdk/aws-codepipeline:crossAccountKeyAliasStackSafeResourceName": true,
+ "@aws-cdk/aws-s3:createDefaultLoggingPolicy": true,
+ "@aws-cdk/aws-sns-subscriptions:restrictSqsDescryption": true,
+ "@aws-cdk/aws-apigateway:disableCloudWatchRole": true,
+ "@aws-cdk/core:enablePartitionLiterals": true,
+ "@aws-cdk/aws-events:eventsTargetQueueSameAccount": true,
+ "@aws-cdk/aws-iam:standardizedServicePrincipals": true,
+ "@aws-cdk/aws-ecs:disableExplicitDeploymentControllerForCircuitBreaker": true,
+ "@aws-cdk/aws-iam:importedRoleStackSafeDefaultPolicyName": true,
+ "@aws-cdk/aws-s3:serverAccessLogsUseBucketPolicy": true,
+ "@aws-cdk/aws-route53-patters:useCertificate": true,
+ "@aws-cdk/customresources:installLatestAwsSdkDefault": false,
+ "@aws-cdk/aws-rds:databaseProxyUniqueResourceName": true,
+ "@aws-cdk/aws-codedeploy:removeAlarmsFromDeploymentGroup": true,
+ "@aws-cdk/aws-apigateway:authorizerChangeDeploymentLogicalId": true,
+ "@aws-cdk/aws-ec2:launchTemplateDefaultUserData": true,
+ "@aws-cdk/aws-secretsmanager:useAttachedSecretResourcePolicyForSecretTargetAttachments": true,
+ "@aws-cdk/aws-redshift:columnId": true,
+ "@aws-cdk/aws-stepfunctions-tasks:enableEmrServicePolicyV2": true,
+ "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": true,
+ "@aws-cdk/aws-apigateway:requestValidatorUniqueId": true,
+ "@aws-cdk/aws-kms:aliasNameRef": true,
+ "@aws-cdk/aws-autoscaling:generateLaunchTemplateInsteadOfLaunchConfig": true,
+ "@aws-cdk/core:includePrefixInUniqueNameGeneration": true,
+ "@aws-cdk/aws-efs:denyAnonymousAccess": true,
+ "@aws-cdk/aws-opensearchservice:enableOpensearchMultiAzWithStandby": true,
+ "@aws-cdk/aws-lambda-nodejs:useLatestRuntimeVersion": true,
+ "@aws-cdk/aws-efs:mountTargetOrderInsensitiveLogicalId": true,
+ "@aws-cdk/aws-rds:auroraClusterChangeScopeOfInstanceParameterGroupWithEachParameters": true,
+ "@aws-cdk/aws-appsync:useArnForSourceApiAssociationIdentifier": true,
+ "@aws-cdk/aws-rds:preventRenderingDeprecatedCredentials": true,
+ "@aws-cdk/aws-codepipeline-actions:useNewDefaultBranchForCodeCommitSource": true,
+ "@aws-cdk/aws-cloudwatch-actions:changeLambdaPermissionLogicalIdForLambdaAction": true,
+ "@aws-cdk/aws-codepipeline:crossAccountKeysDefaultValueToFalse": true
+ }
+}
diff --git a/apigw-dynamodb-python-cdk/example-pattern.json b/apigw-dynamodb-python-cdk/example-pattern.json
new file mode 100644
index 000000000..bbc50fe66
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/example-pattern.json
@@ -0,0 +1,53 @@
+{
+ "title": "API Gateway direct integration to DynamoDB",
+ "description": "Direct integration with API Gateway to DynamoDB with transformation using VTL and CDK and examples for Cognito, Lambda authorizer and API keys.",
+ "language": "Python",
+ "level": "300",
+ "framework": "CDK",
+ "introBox": {
+ "headline": "How it works",
+ "text": [
+ "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."
+ ]
+ },
+ "gitHub": {
+ "template": {
+ "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/apigw-dynamodb-python-cdk",
+ "templateURL": "serverless-patterns/apigw-dynamodb-python-cdk",
+ "projectFolder": "apigw-dynamodb-python-cdk",
+ "templateFile": "apigw_dynamodb_python_cdk_stack.py"
+ }
+ },
+ "resources": {
+ "bullets": [
+ {
+ "text": "API Gateway Integrations",
+ "link": "https://docs.aws.amazon.com/apigateway/latest/developerguide/how-to-integration-settings.html"
+ }
+ ]
+ },
+ "deploy": {
+ "text": [
+ "cdk deploy"
+ ]
+ },
+ "testing": {
+ "text": [
+ "See the GitHub repo for detailed testing instructions."
+ ]
+ },
+ "cleanup": {
+ "text": [
+ "Delete the stack: cdk delete."
+ ]
+ },
+ "authors": [
+ {
+ "name": "Maya Morav Freiman",
+ "image": "https://avatars.githubusercontent.com/u/11615439?v=4",
+ "bio": "Technical Account Manager at AWS",
+ "linkedin": "mayaaws"
+ }
+ ]
+}
diff --git a/apigw-dynamodb-python-cdk/image.png b/apigw-dynamodb-python-cdk/image.png
new file mode 100644
index 000000000..97b7d1df1
Binary files /dev/null and b/apigw-dynamodb-python-cdk/image.png differ
diff --git a/apigw-dynamodb-python-cdk/requirements.txt b/apigw-dynamodb-python-cdk/requirements.txt
new file mode 100644
index 000000000..623c85197
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/requirements.txt
@@ -0,0 +1,5 @@
+aws-cdk-lib==2.130.0
+constructs>=10.0.0,<11.0.0
+pytest==8.2.2
+requests==2.31.0
+boto3
diff --git a/apigw-dynamodb-python-cdk/src/lambda_function.py b/apigw-dynamodb-python-cdk/src/lambda_function.py
new file mode 100644
index 000000000..5612bf38e
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/src/lambda_function.py
@@ -0,0 +1,105 @@
+# A simple request-based authorizer example to demonstrate how to use request
+# parameters to allow or deny a request. In this example, a request is
+# authorized if the client-supplied headerauth1 header, QueryString1
+# query parameter, and stage variable of StageVar1 all match
+# specified values of 'headerValue1', 'queryValue1', and 'stageValue1',
+# respectively.
+
+import json
+import jwt
+import boto3
+
+apigateway_client = boto3.client('apigateway')
+
+def lambda_handler(event, context):
+ # Retrieve request parameters from the Lambda function input:
+ headers = event['headers']
+ # queryStringParameters = event['queryStringParameters']
+ # pathParameters = event['pathParameters']
+ # stageVariables = event['stageVariables']
+
+ # Parse the input for the parameter values
+ tmp = event['methodArn'].split(':')
+ apiGatewayArnTmp = tmp[5].split('/')
+ awsAccountId = tmp[4]
+ region = tmp[3]
+ restApiId = apiGatewayArnTmp[0]
+ stage = apiGatewayArnTmp[1]
+ method = apiGatewayArnTmp[2]
+ resource = '/'
+
+ if (apiGatewayArnTmp[3]):
+ resource += apiGatewayArnTmp[3]
+
+ # Perform authorization to return the Allow policy for correct parameters
+ # and the 'Unauthorized' error, otherwise.
+
+ authResponse = {}
+ condition = {}
+ condition['IpAddress'] = {}
+ id_token = json.dumps(event['headers']['Authorization'])
+ try:
+ token = id_token.split()[1].strip('"')
+ decoded = jwt.decode(token, options={"verify_signature": False})
+ groups = decoded['cognito:groups']
+ for group in groups:
+ if 'Group-' in group:
+ # api_key = get_api_key_value(f'ApiKey-{group.split('-')[1]}')
+ api_key = get_api_key(f'ApiKey-{group.split('-')[1]}')
+ response = generateAllow('me', event['methodArn'], api_key)
+ print('authorized')
+ return json.loads(response)
+ else:
+ print('unauthorized')
+ response = generateDeny('me', event['methodArn'], api_key)
+ return json.loads(response)
+
+ except:
+ token = None
+ api_key = None
+ response = generateDeny('me', event['methodArn'], api_key)
+ return json.loads(response)
+
+
+ # Help function to generate IAM policy
+
+def generatePolicy(principalId, effect, resource, api_key):
+ authResponse = {}
+ authResponse['principalId'] = principalId
+ if (effect and resource):
+ policyDocument = {}
+ policyDocument['Version'] = '2012-10-17'
+ policyDocument['Statement'] = []
+ statementOne = {}
+ statementOne['Action'] = 'execute-api:Invoke'
+ statementOne['Effect'] = effect
+ statementOne['Resource'] = resource
+ policyDocument['Statement'] = [statementOne]
+ authResponse['policyDocument'] = policyDocument
+ authResponse['usageIdentifierKey'] = api_key
+
+
+ authResponse_JSON = json.dumps(authResponse)
+ print(authResponse_JSON)
+
+ return authResponse_JSON
+
+
+def generateAllow(principalId, resource, api_key):
+ return generatePolicy(principalId, 'Allow', resource, api_key)
+
+
+def generateDeny(principalId, resource, api_key):
+ return generatePolicy(principalId, 'Deny', resource, api_key)
+
+
+def get_api_key(key_name):
+ try:
+ response = apigateway_client.get_api_keys(nameQuery=key_name, includeValues=True)
+ api_keys = response['items']
+ if api_keys:
+ return api_keys[0]['value']
+ return None
+ except ClientError as e:
+ print(e)
+ return None
\ No newline at end of file
diff --git a/apigw-dynamodb-python-cdk/tests/__init__.py b/apigw-dynamodb-python-cdk/tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/apigw-dynamodb-python-cdk/tests/test_apigw_dynamodb_python_stack.py b/apigw-dynamodb-python-cdk/tests/test_apigw_dynamodb_python_stack.py
new file mode 100644
index 000000000..54f8f0b88
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/tests/test_apigw_dynamodb_python_stack.py
@@ -0,0 +1,161 @@
+import pytest
+import requests
+import boto3
+from botocore.exceptions import ClientError
+
+def get_stack_outputs(stack_name):
+ client = boto3.client('cloudformation')
+ response = client.describe_stacks(StackName=stack_name)
+ outputs = response['Stacks'][0]['Outputs']
+
+ return {output['OutputKey']: output['OutputValue'] for output in outputs}
+
+@pytest.fixture(scope="module")
+def config():
+ stack_name = "ApigwDynamodbPythonStack"
+ output = get_stack_outputs(stack_name)
+ return output
+
+def get_api_key_value(apiKey):
+ client = boto3.client('apigateway')
+ response = client.get_api_key(
+ apiKey=apiKey,
+ includeValue=True
+ )
+ return response['value']
+
+def create_cognito_user(username, password, client_id, pool_id, group_name):
+ client = boto3.client('cognito-idp')
+
+ try:
+ response = client.admin_create_user(
+ UserPoolId=pool_id,
+ Username=username,
+ UserAttributes=[
+ {
+ 'Name': 'email',
+ 'Value': username
+ }
+ ],
+ # TemporaryPassword=password,
+ MessageAction='SUPPRESS'
+ )
+ response = client.admin_set_user_password(
+ UserPoolId=pool_id,
+ Username=username,
+ Password=password,
+ Permanent=True
+ )
+ client.admin_add_user_to_group(
+ UserPoolId=pool_id,
+ Username=username,
+ GroupName=group_name
+ )
+ return response
+ except ClientError as e:
+ print(f"Error creating user: {e}")
+ return None
+
+# Function to authenticate user and retrieve token
+def authenticate_user(username, password, client_id):
+ client = boto3.client('cognito-idp')
+
+ try:
+ response = client.initiate_auth(
+ ClientId=client_id,
+ AuthFlow='USER_PASSWORD_AUTH',
+ AuthParameters={
+ 'USERNAME': username,
+ 'PASSWORD': password,
+ }
+ )
+ return response['AuthenticationResult']['IdToken']
+ except ClientError as e:
+ print(f"Error authenticating user: {e}")
+ return None
+
+# Function to delete a user from Cognito
+def delete_cognito_user(username, pool_id):
+ client = boto3.client('cognito-idp')
+
+ try:
+ client.admin_delete_user(
+ UserPoolId=pool_id,
+ Username=username
+ )
+ except ClientError as e:
+ print(f"Error deleting user: {e}")
+
+@pytest.fixture(scope="module")
+def token(config):
+ usernameBasic = 'testuser@mail.com' # Choose a unique username
+ usernameFree = "testuser1@mail.com"
+ password = 'TestPassword123!' # Ensure this meets Cognito password policy
+ client_id=config['CognitoClientId']
+ pool_id=config['CognitoUserPoolId']
+ # Create user
+ create_cognito_user(usernameFree, password, client_id, pool_id, "Group-FreeTier")
+ create_cognito_user(usernameBasic, password, client_id, pool_id, "Group-BasicUsagePlan")
+
+ # Authenticate user and get token
+ user_token_free = authenticate_user(usernameFree, password, client_id)
+ user_token_basic = authenticate_user(usernameBasic, password, client_id)
+
+
+ # Finalizer to delete user after tests
+ yield user_token_free, user_token_basic # This is where the test will use the token
+ delete_cognito_user(usernameFree, pool_id) # This will run after the test is done
+ delete_cognito_user(usernameBasic, pool_id)
+
+
+def test_put_authorized(token, config):
+ api_url = config['ApiUrl']
+ api_url = f"{api_url}/put"
+ token_data = token[1]
+ headers = {'Authorization': f'Bearer {token_data}', 'Content-Type': 'application/json'}
+ payload = {"ID": "aa", "FirstName": "test", "Age": 22}
+ response = requests.post(api_url, headers=headers, json=payload)
+
+ assert response.status_code == 200
+ assert response.json()['body'] == {"message": "Item Added Successfully"}
+
+def test_put_unauthorized(config):
+ api_url = config['ApiUrl']
+ api_url = f"{api_url}/put"
+ headers = {'Authorization': f'Bearer invalid_token', 'Content-Type': 'application/json'}
+ payload = {"ID": "aa", "FirstName": "test", "Age": 22}
+
+ response = requests.post(api_url, headers=headers, json=payload)
+
+ assert response.status_code == 403
+
+def test_get_with_api_key(config):
+ api_url = config['ApiUrl']
+ api_url = f"{api_url}/get"
+ api_key_value = get_api_key_value(config['ApiKeyId0'])
+ headers = {'x-api-key': api_key_value, 'Content-Type': 'application/json'}
+
+ response = requests.get(api_url, headers=headers)
+
+ assert response.status_code == 200
+ assert response.json()['body'] == {"ID": "aa", "FirstName": "test", "Age": '22'}
+
+def test_get_without_api_key(config):
+ api_url = config['ApiUrl']
+ api_url = f"{api_url}/get"
+ headers = {'Content-Type': 'application/json'}
+
+ response = requests.get(api_url, headers=headers)
+ print(response.text)
+
+ assert response.status_code == 403
+ assert response.json() == {'message': 'Forbidden'}
+
+def test_delete_item_authorized(token, config):
+ api_url = config['ApiUrl']
+ api_url = f"{api_url}/delete/aa"
+ headers = {'Authorization': f'Bearer {token[0]}', 'Content-Type': 'application/json'}
+ payload = {'ID': {'S': 'aa'}, 'FirstName': {'S': 'test'}}
+
+ response = requests.post(api_url, headers=headers, json=payload)
+ assert response.status_code == 200
diff --git a/apigw-dynamodb-python-cdk/vtl/deleteItem.vtl b/apigw-dynamodb-python-cdk/vtl/deleteItem.vtl
new file mode 100644
index 000000000..fedb6c3d5
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/vtl/deleteItem.vtl
@@ -0,0 +1,8 @@
+#set($inputRoot = $input.path('$'))
+{
+ "TableName": "$stageVariables.TableName",
+ "Key": {
+ "ID": { "S": "$inputRoot.ID.S" },
+ "FirstName": {"S": "$inputRoot.FirstName.S"}
+ }
+}
diff --git a/apigw-dynamodb-python-cdk/vtl/putItem.vtl b/apigw-dynamodb-python-cdk/vtl/putItem.vtl
new file mode 100644
index 000000000..2325cdd58
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/vtl/putItem.vtl
@@ -0,0 +1,14 @@
+#set($inputRoot = $input.path('$'))
+{
+ "TableName": "$stageVariables.TableName",
+ "Item": {
+ "ID": { "S": "$inputRoot.ID" },
+ "FirstName": { "S": "$inputRoot.FirstName" },
+ "Age": { "N": "$inputRoot.Age" }
+ #foreach($key in $inputRoot.keySet())
+ #if ($key != 'FirstName' && $key != 'Age' && $key != 'ID')
+ ,"$key": { "S": "$inputRoot.get($key)" }
+ #end
+ #end
+ }
+}
diff --git a/apigw-dynamodb-python-cdk/vtl/response.vtl b/apigw-dynamodb-python-cdk/vtl/response.vtl
new file mode 100644
index 000000000..d2a6d65bf
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/vtl/response.vtl
@@ -0,0 +1,9 @@
+## Response mapping template for DynamoDB PutItem operation
+#set($inputRoot = $input.path('$'))
+{
+ "statusCode": "200",
+ "body": {"message": "Item Added Successfully"},
+ "headers": {
+ "Content-Type": "application/json"
+ }
+}
diff --git a/apigw-dynamodb-python-cdk/vtl/scan.vtl b/apigw-dynamodb-python-cdk/vtl/scan.vtl
new file mode 100644
index 000000000..bb00f9450
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/vtl/scan.vtl
@@ -0,0 +1,15 @@
+#set($inputRoot = $input.path('$'))
+{
+ "statusCode": "200",
+ "headers": {
+ "Content-Type": "application/json"
+ },
+ "body":
+#foreach($elem in $inputRoot.Items)
+ {
+ "ID" : "$elem.ID.S",
+ "FirstName": "$elem.FirstName.S",
+ "Age" : "$elem.Age.N"
+ }#if($foreach.hasNext),#end
+#end
+}
\ No newline at end of file
diff --git a/apigw-dynamodb-python-cdk/vtl/scan_request.vtl b/apigw-dynamodb-python-cdk/vtl/scan_request.vtl
new file mode 100644
index 000000000..2dd1b6f12
--- /dev/null
+++ b/apigw-dynamodb-python-cdk/vtl/scan_request.vtl
@@ -0,0 +1,3 @@
+{
+ "TableName": "$stageVariables.TableName"
+}
\ No newline at end of file