API - Choices module

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

Example

Typical Usage::

from stacks.resources.api_gateway.resources import choices

choices.create(stack, created_api_gateway, dynamodb_choicess_table_arn, auth, question_updater_function_arn)

__create_get_choice(ddb_role, dynamodb_pkey_id, dynamodb_table, resource, response_model, request_validator, auth)

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

Creates a GET method on the /{model_name}/{id} resource to fetch a single choice from 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.
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/choices.py

def __create_get_choice(ddb_role: Role, dynamodb_pkey_id: str, dynamodb_table: Table, resource: apigateway.Resource, response_model: apigateway.Model, request_validator: apigateway.RequestValidator,  auth: apigateway.CognitoUserPoolsAuthorizer):
    """Create the GET Choice methods on the provided REST API Gateway Resource.

    Creates a GET method on the `/{model_name}/{id}` resource to fetch a single choice from 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.
        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_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)
                            {
                                "choice_id": "$inputRoot.choice_id.S",
                                "choice_lense": "$inputRoot.lense.S",
                                "choice_desc": "$inputRoot.choice_desc.S",
                                "choice_name": "$inputRoot.choice_name.S",
                                "choice_risk": "$inputRoot.choice_risk.S",
                                "choice_pillar": "$inputRoot.pillar_name.S",
                                "choice_question": "$inputRoot.question_title.S",
                                "choice_question_id": "$inputRoot.question_id.S" 
                            }               
                        '''   
                    }
                ),
                __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": response_model}, 
        response_parameters={ 'method.response.header.Access-Control-Allow-Origin': True }
    )


    resource.add_method(
        'GET', 
        integration=get_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_choices(ddb_role, dynamodb_table, resource, response_model, request_validator, auth)

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

Creates a GET method on the /{model_name} resource to fetch all choices from 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.
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/choices.py

def __create_get_choices(ddb_role: Role, dynamodb_table: Table, resource: apigateway.Resource, response_model: apigateway.Model, request_validator: apigateway.RequestValidator,  auth: apigateway.CognitoUserPoolsAuthorizer):
    """Create the GET Choices methods on the provided REST API Gateway Resource.

    Creates a GET method on the `/{model_name}` resource to fetch all choices from 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.
        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_integration = apigateway.AwsIntegration(
        service="dynamodb",
        action="Scan",
        options= apigateway.IntegrationOptions(
            credentials_role=ddb_role,
            passthrough_behavior=apigateway.PassthroughBehavior.NEVER,
            request_parameters={
                "integration.request.querystring.lense": "method.request.querystring.lense",
                "integration.request.querystring.pillar": "method.request.querystring.pillar"
            },  
            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($choice in $inputRoot.Items)
                                {
                                    "choice_id": "$choice.choice_id.S",
                                    "choice_lense": "$choice.lense.S",
                                    "choice_desc": "$choice.choice_desc.S",
                                    "choice_name": "$choice.choice_name.S",
                                    "choice_risk": "$choice.choice_risk.S",
                                    "choice_pillar": "$choice.pillar_name.S",
                                    "choice_question": "$choice.question_title.S",
                                    "choice_question_id": "$choice.question_id.S" 
                                }  
                                #if($foreach.hasNext),#end
                                #end                                       
                            ]
                        ''' 
                    }
                ),
                __ERROR_INTEGRATION_RESPONSE_400,
                __ERROR_INTEGRATION_RESPONSE_500
            ],
            request_templates= {
                'application/json': '''
                    #set($queryIsSet = false)  
                    #if($!{method.request.querystring.lense} != "" && $!{method.request.querystring.pillar} != "")
                        #set($queryIsSet = true)   
                    #end

                    {
                        "TableName": "%s"#if($queryIsSet),#end
                        #if($queryIsSet)
                        "FilterExpression": "lense = :lense AND pillar_name = :pillar",
                        "ExpressionAttributeValues": {
                            ":lense": { "S": "$method.request.querystring.lense" },
                            ":pillar": { "S": "$method.request.querystring.pillar" }
                        }
                        #end
                    }
                    '''  % (dynamodb_table.table_name)
            },
        ),
    )


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

    request_parameters = { 
        'method.request.querystring.lense': False, 
        'method.request.querystring.pillar': False 
    }

    resource.add_method(
        'GET', 
        integration=get_integration, 
        method_responses=[
            success_response_200,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_validator=request_validator,
        request_parameters=request_parameters,
        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 model 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/choices.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 model 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_update_questions(question_updater_function, resource, request_model, response_model, request_validator, auth)

Create the POST Question Update methods on the provided REST API Gateway Resource.

Creates a POST method on the /{model_name}/update resource to trigger the update of all Choices.

Attributes:

Name	Type	Description
question_updater_function	aws_cdk.aws_lambda.Function
resource	aws_cdk.aws_apigateway.Resource	API Gateway resource where the resources and methods should be created.
request_model	aws_cdk.aws_apigateway.Model	Model for the request validation.
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/choices.py

def __create_post_update_questions(
    question_updater_function: Function,
    resource: apigateway.Resource, 
    request_model: apigateway.Model,
    response_model: apigateway.Model,
    request_validator: apigateway.RequestValidator, 
    auth: apigateway.CognitoUserPoolsAuthorizer
    ):
    """Create the POST Question Update methods on the provided REST API Gateway Resource.

    Creates a POST method on the `/{model_name}/update` resource to trigger the update of all Choices.

    Attributes:
        question_updater_function (aws_cdk.aws_lambda.Function): 
        resource (aws_cdk.aws_apigateway.Resource): API Gateway resource where the resources and methods should be created.
        request_model (aws_cdk.aws_apigateway.Model): Model for the request validation.
        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.
    """

    post_integration = apigateway.LambdaIntegration(
        question_updater_function,
        proxy=False,
        passthrough_behavior=apigateway.PassthroughBehavior.NEVER,
        request_parameters={
            "integration.request.header.X-Amz-Invocation-Type": "'Event'",
        },     
        integration_responses=[
                apigateway.IntegrationResponse(
                    status_code="202",
                    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( { } ),
            },        
        )

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

    resource.add_method(
        'POST', 
        integration=post_integration, 
        method_responses=[
            success_response_202,
            __ERROR_METHOD_RESPONSE_400, 
            __ERROR_METHOD_RESPONSE_500
        ],
        request_parameters={
            "method.request.header.X-Amz-Invocation-Type": True
        },        
        request_validator=request_validator,
        request_models={ "application/json": request_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/choices.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, question_updater_function_arn, dynamodb_pkey_id='choice_id', model_name='choices')

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 Choices.
auth	aws_cdk.aws_cognito.CognitoUserPoolsAuthorizer	Cognito User Pool Authorizer used for the endpoint authentication.
question_updater_function_arn	str	ARN of the Lambda function which updates the choices.
dynamodb_pkey_id	Optional[str]	Primary Key of the DynamoDB Table, default choice_id.
model_name	Optional[str]	The path for the resources, default choices.

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

def create(
    stack: core.Stack, 
    rest_api_gateway: apigateway.RestApi, 
    dynamodb_table_arn: str, 
    auth: apigateway.CognitoUserPoolsAuthorizer,
    question_updater_function_arn: str,
    dynamodb_pkey_id: str = "choice_id", 
    model_name: str = "choices") -> 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 Choices.
        auth (aws_cdk.aws_cognito.CognitoUserPoolsAuthorizer): Cognito User Pool Authorizer used for the endpoint authentication.
        question_updater_function_arn (str): ARN of the Lambda function which updates the choices.
        dynamodb_pkey_id (Optional[str]): Primary Key of the DynamoDB Table, default `choice_id`.
        model_name (Optional[str]): The path for the resources, default `choices`.

    """

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

    ddb_get_role = create_integration_iam_role(stack, ["dynamodb:GetItem", "dynamodb:Scan"], [dynamodb_table_arn], "WaReviewToolChoiceDynamoDbGet")

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

    get_choice_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT,
        required=["choice_id", "choice_lense", "choice_pillar", "choice_desc", "choice_question", "choice_question_id"],
        properties={
            "choice_id": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                min_length=4,
                max_length=64,
            ),
            "choice_lense": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                enum=["wellarchitected", "foundationaltechni", "serverless", "softwareasaservice"]
            ),
            "choice_pillar": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                enum=["operationalExcellence", "security", "reliability", "performance", "costOptimization", "sustainability"]
            ),
            "choice_desc": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING
            ),   
            "choice_question": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
            ),
            "choice_question_id": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                min_length=4,
                max_length=64
            ),                           
            "choice_risk": apigateway.JsonSchema(
                type=apigateway.JsonSchemaType.STRING,
                default="NONE",
                enum=["NONE", "MEDIUM", "HIGH"]
            )
        }
    )
    get_choice_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="GetChoiceResponseModel", content="application/json", schema=get_choice_response_model_schema, description="Validate the response body")    

    get_choices_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.ARRAY,
        items=get_choice_response_model_schema
    )

    get_choices_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="GetChoicesResponseModel", content="application/json", schema=get_choices_response_model_schema, description="Validate the response body")    

    post_question_update_request_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT
    )

    post_question_update_response_model_schema = apigateway.JsonSchema(
        type=apigateway.JsonSchemaType.OBJECT
    )

    post_question_update_request_model = __create_model(rest_api_gateway=rest_api_gateway, name="PostQuestionUpdateRequestModel", content="application/json", schema=post_question_update_request_model_schema, description="Validate the request body")    
    post_question_update_response_model = __create_model(rest_api_gateway=rest_api_gateway, name="PostQuestionUpdateResponseModel", content="application/json", schema=post_question_update_response_model_schema, description="Validate the response body")    

    #
    ## Functions
    #     

    question_updater_function = Function.from_function_arn(stack, "WaReviewToolQuestionUpdaterFunction", question_updater_function_arn)

    CfnPermission(
        stack,
        'WaReviewToolQuestionUpdaterFunctionPermission',
        action="lambda:InvokeFunction",
        function_name=question_updater_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="Choices")

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

    __create_get_choices(ddb_role=ddb_get_role, dynamodb_table=dynamodb_table, resource=choices_resource, response_model=get_choices_response_model, request_validator=request_validator, auth=auth)

    choice_resource = choices_resource.add_resource('{id}')
    __create_get_choice(ddb_role=ddb_get_role, dynamodb_pkey_id=dynamodb_pkey_id, dynamodb_table=dynamodb_table, resource=choice_resource, response_model=get_choice_response_model, request_validator=request_validator, auth=auth)

    update_questions_resource = choices_resource.add_resource('update')

    __create_post_update_questions(question_updater_function=question_updater_function, resource=update_questions_resource, request_model=post_question_update_request_model, response_model=post_question_update_response_model, request_validator=request_validator, auth=auth)