API - Customer module

This module creates the Customers resources on a provided AWS API Gateway.

Example

Typical Usage::

from stacks.resources.api_gateway.resources import customers

customers.create(stack, created_api_gateway, dynamodb_customers_table_arn, auth, customer_credentials_function_arn)

__create_delete_customer(ddb_role, dynamodb_pkey_id, dynamodb_table, customer_resource, customer_response_model, request_validator, auth)

Create the DELETE Customer methods on the provided REST API Gateway Resource.

Creates a DELETE method on the /{model_name}/{id} resource to delete a single customer from the DynamoDB. Additionally

Attributes:

Name	Type	Description
ddb_role	aws_cdk.aws_iam.Role	The IAM Role to access the DB from the API Gateway.
dynamodb_pkey_id	str	DynamoDB Table primary key name.
dynamodb_table	aws_cdk.aws_dynamodb.Table	The DynamoDB Table.
customer_resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
customer_response_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
request_validator	aws_cdk.aws_apigateway.RequestValidator	Validator for the request.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_delete_customer(
    ddb_role: Role, 
    dynamodb_pkey_id: str, 
    dynamodb_table: Table, 
    customer_resource: apigateway.Resource, 
    customer_response_model: apigateway.Model,
    request_validator: apigateway.RequestValidator, 
    auth: apigateway.CognitoUserPoolsAuthorizer
    ):
    """Create the DELETE Customer methods on the provided REST API Gateway Resource.

    Creates a DELETE method on the `/{model_name}/{id}` resource to delete a single customer from the DynamoDB.
    Additionally 

    Attributes:
        ddb_role (aws_cdk.aws_iam.Role): The IAM Role to access the DB from the API Gateway.
        dynamodb_pkey_id (str): DynamoDB Table primary key name.
        dynamodb_table (aws_cdk.aws_dynamodb.Table): The DynamoDB Table.
        customer_resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        customer_response_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        request_validator (aws_cdk.aws_apigateway.RequestValidator): Validator for the request.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    delete_customer_integration = apigateway.AwsIntegration(
        service="dynamodb",
        action="DeleteItem",
        options=apigateway.IntegrationOptions(
            credentials_role=ddb_role,
            integration_responses=[
                apigateway.IntegrationResponse(
                    status_code="200",
                    selection_pattern="200",
                    response_parameters={
                        "method.response.header.Access-Control-Allow-Origin": "'*'"
                    },                      
                ),
                __ERROR_INTEGRATION_RESPONSE_400,
                __ERROR_INTEGRATION_RESPONSE_500
            ],
            request_templates={
                'application/json': json.dumps({
                    "Key": {
                        f"{dynamodb_pkey_id}": {
                        "S": "$method.request.path.id"
                        }
                    },
                    "TableName": f"{dynamodb_table.table_name}"
                    }),
            },
        ),
    )    

    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": customer_response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )

    customer_resource.add_method(
        'DELETE', 
        integration=delete_customer_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_delete_customer_credentials(customer_credentials_function, resource, request_model, response_model, request_validator, auth)

Create the DELETE Customer Credentials methods on the provided REST API Gateway Resource.

Creates a DELETE method on the /{model_name}/{customer_id}/credentials resource to delete credentials for a customer in the Secrets Manager via a Lambda function.

Attributes:

Name	Type	Description
customer_credentials_function	aws_cdk.aws_lambda.Function	The lambda function for the integration.
resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
response_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_delete_customer_credentials( customer_credentials_function: Function, resource: apigateway.Resource, request_model: apigateway.Model, response_model: apigateway.Model, request_validator: apigateway.RequestValidator,  auth: apigateway.CognitoUserPoolsAuthorizer):
    """Create the DELETE Customer Credentials methods on the provided REST API Gateway Resource.

    Creates a DELETE method on the `/{model_name}/{customer_id}/credentials` resource to delete credentials for a customer in the Secrets Manager via a Lambda function.

    Attributes:
        customer_credentials_function (aws_cdk.aws_lambda.Function): The lambda function for the integration.
        resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        response_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    delete_customer_credentials_integration = apigateway.LambdaIntegration(
            customer_credentials_function,
            proxy=False,
            integration_responses=[
                    apigateway.IntegrationResponse(
                        status_code="200",
                        response_parameters={
                            "method.response.header.Access-Control-Allow-Origin": "'*'"
                        },       
                    ),
                    apigateway.IntegrationResponse(
                        selection_pattern=".*\"failure\".*",
                        status_code="500",
                        response_templates= {
                            "application/json": '{"error": "Internal Service Error"}'
                        }
                    )            ],
            request_templates={
                    'application/json': json.dumps(
                        { 
                            "action": "DELETE",
                            "customer_id": "$method.request.path.id"
                        }
                    ),
                },
        )        


    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )

    resource.add_method(
        'DELETE', 
        integration=delete_customer_credentials_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        request_models={ "application/json": request_model }, 
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_get_customer(ddb_role, dynamodb_pkey_id, dynamodb_table, customer_resource, customer_response_model, request_validator, auth)

Create the GET Customer methods on the provided REST API Gateway Resource.

Creates a GET method on the /{model_name} resource to fetch a single customer from the DynamoDB. Additionally

Attributes:

Name	Type	Description
ddb_role	aws_cdk.aws_iam.Role	The IAM Role to access the DB from the API Gateway.
dynamodb_pkey_id	str	DynamoDB Table primary key name.
dynamodb_table	aws_cdk.aws_dynamodb.Table	The DynamoDB Table.
customer_resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
customer_response_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_get_customer(ddb_role: Role, dynamodb_pkey_id: str, dynamodb_table: Table, customer_resource: apigateway.Resource, customer_response_model: apigateway.Model, request_validator: apigateway.RequestValidator,  auth: apigateway.CognitoUserPoolsAuthorizer):
    """Create the GET Customer methods on the provided REST API Gateway Resource.

    Creates a GET method on the `/{model_name}` resource to fetch a single customer from the DynamoDB.
    Additionally 

    Attributes:
        ddb_role (aws_cdk.aws_iam.Role): The IAM Role to access the DB from the API Gateway.
        dynamodb_pkey_id (str): DynamoDB Table primary key name.
        dynamodb_table (aws_cdk.aws_dynamodb.Table): The DynamoDB Table.
        customer_resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        customer_response_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    get_customer_integration = apigateway.AwsIntegration(
        service="dynamodb",
        action="GetItem",
        options= apigateway.IntegrationOptions(
            credentials_role=ddb_role,
            passthrough_behavior=apigateway.PassthroughBehavior.NEVER,
            integration_responses=[
                apigateway.IntegrationResponse(
                    status_code="200",
                    selection_pattern="200",
                    response_parameters={
                        "method.response.header.Access-Control-Allow-Origin": "'*'"
                    },                    
                    response_templates= {
                        "application/json": '''
                            #set($inputRoot = $input.path('$').Item)
                            {
                                "customer_id": "$inputRoot.customer_id.S",
                                "customer_name": "$inputRoot.customer_name.S",
                                "customer_author": {
                                    #foreach($key in $inputRoot.customer_author.M.keySet()) 
                                        "$key": "$inputRoot.customer_author.M[$key].S"
                                    #if($foreach.hasNext),#end
                                #end
                                },
                                "customer_editor": [
                                    #foreach($elem in $inputRoot.customer_editor.L) 
                                        {
                                        #foreach($key in $elem.M.keySet()) 
                                            "$key": "$elem.M[$key].S"#if($foreach.hasNext),#end
                                        #end                                        
                                        }                                        
                                    #if($foreach.hasNext),#end
                                #end
                                ]                              
                            }               
                        '''
                    }
                ),
                __ERROR_INTEGRATION_RESPONSE_400,
                __ERROR_INTEGRATION_RESPONSE_500
            ],
            request_templates= {
                'application/json': json.dumps({
                    "Key": {
                        dynamodb_pkey_id: {
                        "S": "$method.request.path.id"
                        }
                    },
                    "TableName": dynamodb_table.table_name
                    }),
            },
        ),
    )


    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": customer_response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )


    customer_resource.add_method(
        'GET', 
        integration=get_customer_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_get_customer_credentials(customer_credentials_function, resource, response_model, request_validator, auth)

Create the GET Customer Credentials methods on the provided REST API Gateway Resource.

Creates a GET method on the /{model_name}/{customer_id}/credentials resource to fetch credentials for a customer from the Secrets Manager via a Lambda function.

Attributes:

Name	Type	Description
customer_credentials_function	aws_cdk.aws_lambda.Function	The lambda function for the integration.
resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
response_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_get_customer_credentials( customer_credentials_function: Function, resource: apigateway.Resource, response_model: apigateway.Model, request_validator: apigateway.RequestValidator,  auth: apigateway.CognitoUserPoolsAuthorizer):
    """Create the GET Customer Credentials methods on the provided REST API Gateway Resource.

    Creates a GET method on the `/{model_name}/{customer_id}/credentials` resource to fetch credentials for a customer from the Secrets Manager via a Lambda function.

    Attributes:
        customer_credentials_function (aws_cdk.aws_lambda.Function): The lambda function for the integration.
        resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        response_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    get_customer_credentials_integration = apigateway.LambdaIntegration(
            customer_credentials_function,
            proxy=False,
            integration_responses=[
                    apigateway.IntegrationResponse(
                        status_code="200",
                        response_parameters={
                            "method.response.header.Access-Control-Allow-Origin": "'*'"
                        },       
                    ),
                    apigateway.IntegrationResponse(
                        selection_pattern=".*\"failure\".*",
                        status_code="500",
                        response_templates= {
                            "application/json": '{"error": "Internal Service Error"}'
                        }
                    )            ],
            request_templates={
                    'application/json': json.dumps(
                        { 
                            "action": "GET",
                            "customer_id": "$method.request.path.id"
                        }
                    ),
                },
        )

    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )

    resource.add_method(
        'GET', 
        integration=get_customer_credentials_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_get_customers(ddb_role, dynamodb_table, customer_resource, customer_response_model, request_validator, auth)

Create the GET Customer methods on the provided REST API Gateway Resource.

Creates a GET method on the / resource to fetch a all customers from the DynamoDB. Additionally

Attributes:

Name	Type	Description
ddb_role	aws_cdk.aws_iam.Role	The IAM Role to access the DB from the API Gateway.
dynamodb_table	aws_cdk.aws_dynamodb.Table	The DynamoDB Table.
customer_resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
customer_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_get_customers(
    ddb_role: Role, 
    dynamodb_table: Table, 
    customer_resource: apigateway.Resource, 
    customer_response_model: apigateway.Model,
    request_validator: apigateway.RequestValidator, 
    auth: apigateway.CognitoUserPoolsAuthorizer
    ):    
    """Create the GET Customer methods on the provided REST API Gateway Resource.

    Creates a GET method on the `/` resource to fetch a all customers from the DynamoDB.
    Additionally 

    Attributes:
        ddb_role (aws_cdk.aws_iam.Role): The IAM Role to access the DB from the API Gateway.
        dynamodb_table (aws_cdk.aws_dynamodb.Table): The DynamoDB Table.
        customer_resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        customer_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    get_customers_integration = apigateway.AwsIntegration(
        service="dynamodb",
        action="Scan",
        options= apigateway.IntegrationOptions(
            credentials_role=ddb_role,
            integration_responses=[
                apigateway.IntegrationResponse(
                    status_code="200",
                    selection_pattern="200",
                    response_parameters={
                        "method.response.header.Access-Control-Allow-Origin": "'*'"
                    },                      
                    response_templates= {
                        "application/json": '''
                            #set($inputRoot = $input.path('$'))
                            [
                                #foreach($customer in $inputRoot.Items)
                                {
                                    "customer_id": "$customer.customer_id.S",
                                    "customer_name": "$customer.customer_name.S",
                                    "customer_author": {
                                        #foreach($key in $customer.customer_author.M.keySet()) 
                                            "$key": "$customer.customer_author.M[$key].S"
                                        #if($foreach.hasNext),#end
                                    #end                                
                                    },
                                    "customer_editor": [
                                        #foreach($elem in $customer.customer_editor.L) 
                                            {
                                            #foreach($key in $elem.M.keySet()) 
                                                "$key": "$elem.M[$key].S"#if($foreach.hasNext),#end
                                            #end                                        
                                            }

                                        #if($foreach.hasNext),#end
                                    #end
                                    ]  
                                }  
                                #if($foreach.hasNext),#end
                            #end                                       
                            ]
                        '''
                    }                    
                ),
                __ERROR_INTEGRATION_RESPONSE_400,
                __ERROR_INTEGRATION_RESPONSE_500
            ],
            request_templates= {
                'application/json': json.dumps({
                        "TableName": dynamodb_table.table_name
                    }),
            },
        ),
    )    

    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": customer_response_model }, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )


    customer_resource.add_method(
        'GET', 
        integration=get_customers_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_model(rest_api_gateway, name, content, schema, description='')

Helper method to create a model.

Attributes:

Name	Type	Description
rest_api_gateway	aws_cdk.aws_apigateway.RestApi	API Gateway the request validator belongs to.
name	str	Name of the model.
content	str	Content of the schema, e.g. application/json
schema	aws_cdk.aws_apigateway.JsonSchema	Schema for the model.
description	Optional[str]	Description of the model.

Returns:

Type	Description
	aws_cdk.aws_apigateway.Model: The created model.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_model(rest_api_gateway: apigateway.RestApi, name: str, content: str, schema: apigateway.JsonSchema, description: Optional[str] = ""):
    """Helper method to create a model.

    Attributes:
        rest_api_gateway (aws_cdk.aws_apigateway.RestApi): API Gateway the request validator belongs to.
        name (str): Name of the model.
        content (str): Content of the schema, e.g. `application/json`
        schema (aws_cdk.aws_apigateway.JsonSchema): Schema for the model.
        description (Optional[str]): Description of the model.

    Returns:
        aws_cdk.aws_apigateway.Model: The created model.
    """

    model_id: str = "WaReviewTool" + name

    model = rest_api_gateway.add_model(
        model_id,
        content_type=content,
        description=description,
        model_name=name,
        schema=schema
    )

    return model

__create_post_customer(ddb_role, dynamodb_pkey_id, dynamodb_table, customer_resource, customer_model, customer_response_model, request_validator, auth)

Create the POST Customer methods on the provided REST API Gateway Resource.

Creates a POST method on the /{model_name} resource to create a customer in the DynamoDB.

Attributes:

Name	Type	Description
ddb_role	aws_cdk.aws_iam.Role	The IAM Role to access the DB from the API Gateway.
dynamodb_pkey_id	str	DynamoDB Table primary key name.
dynamodb_table	aws_cdk.aws_dynamodb.Table	The DynamoDB Table.
customer_resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
customer_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
request_validator	aws_cdk.aws_apigateway.RequestValidator	Validator for the request.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_post_customer(
    ddb_role: Role, 
    dynamodb_pkey_id: str, 
    dynamodb_table: Table, 
    customer_resource: apigateway.Resource, 
    customer_model: apigateway.Model, 
    customer_response_model: apigateway.Model,
    request_validator: apigateway.RequestValidator, 
    auth: apigateway.CognitoUserPoolsAuthorizer
    ):
    """Create the POST Customer methods on the provided REST API Gateway Resource.

    Creates a POST method on the `/{model_name}` resource to create a customer in the DynamoDB.

    Attributes:
        ddb_role (aws_cdk.aws_iam.Role): The IAM Role to access the DB from the API Gateway.
        dynamodb_pkey_id (str): DynamoDB Table primary key name.
        dynamodb_table (aws_cdk.aws_dynamodb.Table): The DynamoDB Table.
        customer_resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        customer_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        request_validator (aws_cdk.aws_apigateway.RequestValidator): Validator for the request.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    post_customer_integration = apigateway.AwsIntegration(
            service="dynamodb",
            action="PutItem", 
            options=apigateway.IntegrationOptions(
                credentials_role=ddb_role,
                integration_responses=[
                    apigateway.IntegrationResponse(
                        status_code="200",
                        selection_pattern="200",
                        response_parameters={
                            "method.response.header.Access-Control-Allow-Origin": "'*'"
                        },                      
                        response_templates= {
                            "application/json": '{ "customer_id": "$context.requestId" }'
                        }
                    ),
                    __ERROR_INTEGRATION_RESPONSE_400,
                    __ERROR_INTEGRATION_RESPONSE_500
                ],
                passthrough_behavior=apigateway.PassthroughBehavior.NEVER,
                request_templates= {
                    'application/json': '''
                        {
                            "Item": {
                                "%s": {
                                    "S": "$context.requestId"
                                    },
                                "customer_name": { "S": "$input.path('$.customer_name')" },
                                "customer_author": { "M": 
                                    {
                                        "customer_author_name": {
                                            "S": "$input.path('$.customer_author.customer_author_name')"
                                        },
                                        "customer_author_date": {
                                            "S": "$input.path('$.customer_author.customer_author_date')"
                                        }                                
                                    }
                                },
                                "customer_editor": { "L": [ ] }
                            },
                            "TableName": "%s"
                        }
                    ''' % (dynamodb_pkey_id, dynamodb_table.table_name)
                },
            ),
        )

    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": customer_response_model }, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )

    customer_resource.add_method(
        'POST', 
        integration=post_customer_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        request_models={ "application/json": customer_model }, 
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_post_customer_credentials(customer_credentials_function, resource, request_model, response_model, request_validator, auth)

Create the POST Customer Credentials methods on the provided REST API Gateway Resource.

Creates a POST method on the /{model_name}/{customer_id}/credentials resource to create credentials for a customer in the Secrets Manager via a Lambda function.

Attributes:

Name	Type	Description
customer_credentials_function	aws_cdk.aws_lambda.Function	The lambda function for the integration.
resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
response_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_post_customer_credentials( customer_credentials_function: Function, resource: apigateway.Resource, request_model: apigateway.Model, response_model: apigateway.Model, request_validator: apigateway.RequestValidator,  auth: apigateway.CognitoUserPoolsAuthorizer):
    """Create the POST Customer Credentials methods on the provided REST API Gateway Resource.

    Creates a POST method on the `/{model_name}/{customer_id}/credentials` resource to create credentials for a customer in the Secrets Manager via a Lambda function.

    Attributes:
        customer_credentials_function (aws_cdk.aws_lambda.Function): The lambda function for the integration.
        resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        response_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    post_customer_credentials_integration = apigateway.LambdaIntegration(
            customer_credentials_function,
            proxy=False,
            integration_responses=[
                    apigateway.IntegrationResponse(
                        status_code="200",
                        response_parameters={
                            "method.response.header.Access-Control-Allow-Origin": "'*'"
                        },       
                    ),
                    apigateway.IntegrationResponse(
                        selection_pattern=".*\"failure\".*",
                        status_code="500",
                        response_templates= {
                            "application/json": '{"error": "Internal Service Error"}'
                        }
                    )            ],
            request_templates={
                    'application/json': json.dumps(
                        { 
                            "action": "POST",
                            "customer_id": "$method.request.path.id"
                        }
                    ),
                },
        )


    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )

    resource.add_method(
        'POST', 
        integration=post_customer_credentials_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        request_models={ "application/json": request_model }, 
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_put_customer(ddb_role, dynamodb_pkey_id, dynamodb_table, customer_resource, customer_model, customer_response_model, request_validator, auth)

Create the PUT Customer methods on the provided REST API Gateway Resource.

Creates a PUT method on the /{model_name}/{id} resource to update a single customer in the DynamoDB.

Attributes:

Name	Type	Description
ddb_role	aws_cdk.aws_iam.Role	The IAM Role to access the DB from the API Gateway.
dynamodb_pkey_id	str	DynamoDB Table primary key name.
dynamodb_table	aws_cdk.aws_dynamodb.Table	The DynamoDB Table.
customer_resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
customer_model	aws_cdk.aws_apigateway.Model	Model for the response validation.
request_validator	aws_cdk.aws_apigateway.RequestValidator	Validator for the request.
auth	aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer	Authorization for the endpoint.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_put_customer(
    ddb_role: Role, 
    dynamodb_pkey_id: str, 
    dynamodb_table: Table, 
    customer_resource: apigateway.Resource, 
    customer_model: apigateway.Model, 
    customer_response_model: apigateway.Model,
    request_validator: apigateway.RequestValidator, 
    auth: apigateway.CognitoUserPoolsAuthorizer
    ):
    """Create the PUT Customer methods on the provided REST API Gateway Resource.

    Creates a PUT method on the `/{model_name}/{id}` resource to update a single customer in the DynamoDB.

    Attributes:
        ddb_role (aws_cdk.aws_iam.Role): The IAM Role to access the DB from the API Gateway.
        dynamodb_pkey_id (str): DynamoDB Table primary key name.
        dynamodb_table (aws_cdk.aws_dynamodb.Table): The DynamoDB Table.
        customer_resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        customer_model (aws_cdk.aws_apigateway.Model): Model for the response validation.
        request_validator (aws_cdk.aws_apigateway.RequestValidator): Validator for the request.
        auth (aws_cdk.aws_apigateway.CognitoUserPoolsAuthorizer): Authorization for the endpoint.
    """

    put_customer_integration = apigateway.AwsIntegration(
        service="dynamodb",
        action="UpdateItem",
        options= apigateway.IntegrationOptions(
            credentials_role=ddb_role,
            integration_responses=[
                apigateway.IntegrationResponse(
                    status_code="200",
                    selection_pattern="200",
                    response_parameters={
                        "method.response.header.Access-Control-Allow-Origin": "'*'"
                    },                      
                ),
                __ERROR_INTEGRATION_RESPONSE_400,
                __ERROR_INTEGRATION_RESPONSE_500
            ],
            passthrough_behavior=apigateway.PassthroughBehavior.NEVER,
            request_templates= {
                'application/json': '''
                    #set($inputRoot = $input.path('$'))
                    {
                        "TableName": "%s",
                        "Key": { 
                            "%s": { 
                                "S": "$method.request.path.id"
                            } 
                        },
                        "UpdateExpression": "SET customer_name = :name, customer_editor = list_append(customer_editor, :editor)",
                        "ExpressionAttributeValues": {
                            ":name": { "S": "$input.path('$.customer_name')" },
                            ":editor": { "L":  [
                                    { 
                                        "M": 
                                            {
                                            #foreach($key in $inputRoot.customer_author.keySet())   
                                            "$key": {
                                                "S": "$inputRoot.customer_author[$key]" 
                                            }#if($foreach.hasNext),#end
                                            #end                        
                                            }
                                    }
                                ]
                            }          
                        }    
                    }                                                    
                ''' % (dynamodb_table.table_name, dynamodb_pkey_id)
            },
        ),
    )

    success_response_200 = apigateway.MethodResponse(
        status_code='200',
        response_models={ "application/json": customer_response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )

    customer_resource.add_method(
        'PUT', 
        integration=put_customer_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        request_models={ "application/json": customer_model }, 
        authorizer=auth, 
        authorization_type=apigateway.AuthorizationType.COGNITO
    )

__create_request_validator(stack, rest_api_gateway, name)

Helper method to create a request validator.

Attributes:

Name	Type	Description
stack	aws_cdk.aws_core.Stack	AWS CDK Stack to which it is deployed.
rest_api_gateway	aws_cdk.aws_apigateway.RestApi	API Gateway the request validator belongs to.
name	str	Name of the request validator.

Returns:

Type	Description
apigateway.RequestValidator	aws_cdk.aws_apigateway.RequestValidator: The created request validator.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def __create_request_validator(stack: core.Stack, rest_api_gateway: apigateway.RestApi, name: str) -> apigateway.RequestValidator:
    """Helper method to create a request validator.

    Attributes:
        stack (aws_cdk.aws_core.Stack): AWS CDK Stack to which it is deployed.
        rest_api_gateway (aws_cdk.aws_apigateway.RestApi): API Gateway the request validator belongs to.
        name (str): Name of the request validator.

    Returns:
        aws_cdk.aws_apigateway.RequestValidator: The created request validator.
    """

    request_validator_id = "WaReview" + name + "RequestValidator"

    request_validator = apigateway.RequestValidator(
            stack,
            request_validator_id,
            rest_api=rest_api_gateway, 
            validate_request_body=True,
            validate_request_parameters=False
        )    
    return request_validator

create(stack, rest_api_gateway, dynamodb_table_arn, auth, customer_credentials_function_arn, dynamodb_pkey_id='customer_id', model_name='customers')

Create the API resource and methods on the provided REST API Gateway.

Attributes:

Name	Type	Description
stack	aws_cdk.core.Stack	CDK Stack.
rest_api_gateway	aws_cdk.aws_apigateway.RestApi	Rest API to add to.
dynamodb_table_arn	str	ARN of the DynamoDB Tabkle containing the Customers.
auth	aws_cdk.aws_cognito.CognitoUserPoolsAuthorizer	Cognito User Pool Authorizer used for the endpoint authentication.
customer_credentials_function_arn	str	ARN of the Lambda function to fetch & update the customer credentials.
dynamodb_pkey_id	Optional[str]	Primary Key of the DynamoDB Table, default customer_id.
model_name	Optional[str]	The path for the resources, default customers.

Source code in infrastructure/stacks/resources/api_gateway/resources/customers.py

def create(
    stack: core.Stack, 
    rest_api_gateway: apigateway.RestApi, 
    dynamodb_table_arn: str, 
    auth: apigateway.CognitoUserPoolsAuthorizer,
    customer_credentials_function_arn: str,
    dynamodb_pkey_id: str = "customer_id", 
    model_name: str = "customers") -> None:
    """Create the API resource and methods on the provided REST API Gateway.

    Attributes:
        stack (aws_cdk.core.Stack): CDK Stack.        
        rest_api_gateway (aws_cdk.aws_apigateway.RestApi): Rest API to add to.
        dynamodb_table_arn (str): ARN of the DynamoDB Tabkle containing the Customers.
        auth (aws_cdk.aws_cognito.CognitoUserPoolsAuthorizer): Cognito User Pool Authorizer used for the endpoint authentication.
        customer_credentials_function_arn (str): ARN of the Lambda function to fetch & update the customer credentials.
        dynamodb_pkey_id (Optional[str]): Primary Key of the DynamoDB Table, default `customer_id`.
        model_name (Optional[str]): The path for the resources, default `customers`.

    """

    dynamodb_table = Table.from_table_arn(stack, "WaReviewToolCustomersDynamoDbTable", dynamodb_table_arn)

    ddb_write_role = create_integration_iam_role(stack, ["dynamodb:UpdateItem", "dynamodb:PutItem"], [dynamodb_table_arn], "WaReviewCustomerDynamoDbPut")
    ddb_get_role = create_integration_iam_role(stack, ["dynamodb:GetItem", "dynamodb:Scan"], [dynamodb_table_arn], "WaReviewCustomerDynamoDbGet")
    ddb_delete_role = create_integration_iam_role(stack, ["dynamodb:DeleteItem"], [dynamodb_table_arn], "WaReviewCustomerDynamoDbDelete")


    ##########
    # Models
    ##########


    author_editor_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_author_name","customer_author_date"],
        properties={
            "customer_author_name": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                min_length=4,
                max_length=64,
                pattern="^([a-zA-Z0-9]){0,}$"
                ),
            "customer_author_date": apigateway.JsonSchema(type=apigateway.JsonSchemaType.STRING)
        },
    )

    get_customer_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_name", "customer_author"],
        properties={
            "customer_id": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                pattern="^([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{10,14})$"            
                ),
            "customer_name": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                min_length=4,
                max_length=64,
                pattern="^([a-zA-Z0-9- _\.&!#=()\s]){0,}$"
                ),
            "customer_author": author_editor_schema,
            "customer_editor": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.ARRAY,
                items=author_editor_schema
            )
        }
    )
    get_customer_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="GetCustomerResponseModel", content="application/json", schema=get_customer_response_model_schema, description="Validate the Customer Response body")    

    get_customers_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.ARRAY,
        items=[ 
            apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.OBJECT,
                required=["customer_id", "customer_name", "customer_author", "customer_editor"],
                properties={
                    "customer_id": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING,
                        pattern="^([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{10,14})$"            
                        ),
                    "customer_name": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING,
                        min_length=4,
                        max_length=64,
                        pattern="^([a-zA-Z0-9- _\.&!#=()\s]){0,}$"
                        ),
                    "customer_author": author_editor_schema,
                    "customer_editor": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.ARRAY,
                        items=author_editor_schema
                    )
                }
            )
        ]
    )
    get_customers_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="GetCustomersResponseModel", content="application/json", schema=get_customers_response_model_schema, description="Validate the Customer Response body")    

    put_customer_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_name", "customer_author"],
        properties={
            "customer_id": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                pattern="^([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{10,14})$"            
                ),
            "customer_name": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                min_length=4,
                max_length=64,
                pattern="^([a-zA-Z0-9- _\.&!#=()\s]){0,}$"
                ),
            "customer_author": author_editor_schema,
            "customer_editor": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.ARRAY,
                items=author_editor_schema
            )
        }
    )

    put_customer_model = __create_model(rest_api_gateway=rest_api_gateway, name="PutCustomerModel", content="application/json", schema=put_customer_model_schema, description="Validate the Customer Request body")    
    put_customer_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="PutCustomerResponseModel", content="application/json", schema=apigateway.JsonSchema(type=apigateway.JsonSchemaType.OBJECT,properties={}), description="Validate the Customer Response body")    

    post_customer_request_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_name", "customer_author"],
        properties={
            "customer_id": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                pattern="^([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{10,14})$"            
                ),
            "customer_name": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                min_length=4,
                max_length=64,
                pattern="^([a-zA-Z0-9- _\.&!#=()\s]){0,}$"
                ),
            "customer_author": author_editor_schema,
            "customer_editor": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.ARRAY,
                items=author_editor_schema
            )
        }
    )

    post_customer_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_id"],
        properties={
            "customer_id": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                pattern="^([a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{10,14})$"            
            ),        
        }
    )

    post_customer_request_model = __create_model(rest_api_gateway=rest_api_gateway, name="PostCustomerRequestModel", content="application/json", schema=post_customer_request_model_schema, description="Validate the Customer Request body")    
    post_customer_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="PostCustomerResponseModel", content="application/json", schema=post_customer_response_model_schema, description="Validate the Customer Response body")    

    delete_customer_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="DeleteCustomerResponseModel", content="application/json", schema=apigateway.JsonSchema(type=apigateway.JsonSchemaType.OBJECT,properties={}), description="Validate the Customer Response body")    

    get_customer_credentials_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_name", "customer_author"],
        properties={
            "statusCode": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.NUMBER,
                pattern="^[1-5][0-9][0-9]$"
            ),
            "headers": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.OBJECT,
                properties={
                    "Content-Type": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    ),
                    "Access-Control-Allow-Origin": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    ),                     
                }
            ),
            "body": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.OBJECT,
                properties={
                    "username": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    ),
                    "password": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    )
                }
            ),
            "isBase64Encoded": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.BOOLEAN
            )
        }
    )

    get_customer_credentials_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="GetCustomerCredentialsResponseModel", content="application/json", schema=get_customer_credentials_response_model_schema, description="Validate the Customer Credentials Response body")    

    post_customer_credentials_request_model_schema = apigateway.JsonSchema(type=apigateway.JsonSchemaType.OBJECT,properties={})
    post_customer_credentials_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["customer_name", "customer_author"],
        properties={
            "statusCode": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.NUMBER,
                pattern="^[1-5][0-9][0-9]$"
            ),
            "headers": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.OBJECT,
                properties={
                    "Content-Type": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    ),
                    "Access-Control-Allow-Origin": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    ),                     
                }
            ),
            "body": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.OBJECT,
                properties={
                    "username": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    ),
                    "password": apigateway.JsonSchema(
                        type=apigateway.JsonSchemaType.STRING
                    )
                }
            ),
            "isBase64Encoded": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.BOOLEAN
            )
        }
    )
    post_customer_credentials_request_model = __create_model(rest_api_gateway=rest_api_gateway, name="PostCustomerCredentialsRequestModel", content="application/json", schema=post_customer_credentials_request_model_schema, description="Validate the Customer Credentials Response body")    
    post_customer_credentials_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="PostCustomerCredentialsResponseModel", content="application/json", schema=post_customer_credentials_response_model_schema, description="Validate the Customer Credentials Response body")    

    delete_customer_credentials_request_model = __create_model(rest_api_gateway=rest_api_gateway, name="DeleteCustomerCredentialsRequestModel", content="application/json", schema=apigateway.JsonSchema(type=apigateway.JsonSchemaType.OBJECT,properties={}), description="Validate the Customer Credentials Request body")    
    delete_customer_credentials_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="DeleteCustomerCredentialsResponseModel", content="application/json", schema=apigateway.JsonSchema(type=apigateway.JsonSchemaType.OBJECT,properties={}), description="Validate the Customer Credentials Response body")    

    #
    ## Functions
    #
    customer_credentials_function = Function.from_function_arn(stack, "WaReviewToolCustomerCredentialsFunction", customer_credentials_function_arn)

    CfnPermission(
        stack,
        'WaReviewToolCustomerCredentialsFunctionPermission',
        action="lambda:InvokeFunction",
        function_name=customer_credentials_function_arn,
        principal='apigateway.amazonaws.com',
        source_arn=rest_api_gateway.arn_for_execute_api() 
    )

    #
    ## Resources and Methods
    #
    request_validator = __create_request_validator(stack=stack, rest_api_gateway=rest_api_gateway, name="Customer")

    customers_resource = rest_api_gateway.root.add_resource(model_name.lower())

    __create_post_customer(ddb_role=ddb_write_role, dynamodb_pkey_id=dynamodb_pkey_id, dynamodb_table=dynamodb_table, customer_resource=customers_resource, customer_model=post_customer_request_model, customer_response_model=post_customer_response_model, request_validator=request_validator, auth=auth)

    __create_get_customers(ddb_role=ddb_get_role, dynamodb_table=dynamodb_table, customer_resource=customers_resource, customer_response_model=get_customers_response_model, request_validator=request_validator, auth=auth)

    customer_resource = customers_resource.add_resource('{id}')
    __create_get_customer(ddb_role=ddb_get_role, dynamodb_pkey_id=dynamodb_pkey_id, dynamodb_table=dynamodb_table, customer_resource=customer_resource, customer_response_model=get_customer_response_model, request_validator=request_validator, auth=auth)

    __create_put_customer(ddb_role=ddb_write_role, dynamodb_pkey_id=dynamodb_pkey_id, dynamodb_table=dynamodb_table, customer_resource=customer_resource, customer_model=put_customer_model, customer_response_model=put_customer_response_model, request_validator=request_validator, auth=auth)

    __create_delete_customer(ddb_role=ddb_delete_role, dynamodb_pkey_id=dynamodb_pkey_id, dynamodb_table=dynamodb_table, customer_resource=customer_resource, customer_response_model=delete_customer_response_model, request_validator=request_validator, auth=auth)

    customer_credentials_resource = customer_resource.add_resource('credentials')

    __create_get_customer_credentials(customer_credentials_function=customer_credentials_function, resource=customer_credentials_resource, response_model=get_customer_credentials_response_model, request_validator=request_validator, auth=auth)

    __create_post_customer_credentials(customer_credentials_function=customer_credentials_function, resource=customer_credentials_resource, request_model=post_customer_credentials_request_model, response_model=post_customer_credentials_response_model, request_validator=request_validator, auth=auth)