Amazon Q 从入门到精通 – 加速构建亚马逊云基础设施

create a vpc, meeting below requirement:- named "Q-VPC"- cidr：10.0.0.0/16- in each AZ, there're 4 subnets.  - one subnet for jumpserver, the prefix is "pub_", 64 ips in total.  - one subnet for db, the prefix is "db_", 256 ips in total.  - one subnet for middleware, the the prefix is "mw_", 256 ips in total  - one subnet for workload, the prefix is "wk_", 8192 ips in totial.- totally 3 AZs.- only one natgateway.- a python based cdk project.

create a cloudfront, meeting below requirement:- a S3 bucket as origin- there's a test text file named abc.txt, being put in s3 bucket- print full test url in the end of cdk deploy.

requirements:I need a suite of serverless application, includes:1) apigateway, provides restapi, integration with lambda, requires integration with cognito.2) a lambda function base on docker, as backend.3) dynamodb is the storage.4) a cdk project to deploy all of above components.please create full framework code, in a well done project structure.

from aws_cdk import (    Stack,    aws_apigateway as apigateway,    aws_lambda as lambda_,    aws_dynamodb as dynamodb,    aws_cognito as cognito,    aws_iam as iam,    Duration,    RemovalPolicy)from constructs import Construct

class ApiStack(Stack):    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:        super().__init__(scope, construct_id, **kwargs)
        # Create Cognito User Pool        user_pool = cognito.UserPool(            self, "UserPool",            self_sign_up_enabled=True,            auto_verify=cognito.AutoVerifiedAttrs(email=True),            standard_attributes=cognito.StandardAttributes(                email=cognito.StandardAttribute(required=True, mutable=True)            )        )
        # Create Cognito App Client        client = user_pool.add_client("app-client",            auth_flows=cognito.AuthFlow(user_password=True)        )
        # Create DynamoDB table        table = dynamodb.Table(            self, "ItemsTable",            partition_key=dynamodb.Attribute(                name="id",                type=dynamodb.AttributeType.STRING            ),            removal_policy=RemovalPolicy.DESTROY,            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST        )
        # Create Lambda function using Docker        backend_function = lambda_.DockerImageFunction(            self, "BackendFunction",            code=lambda_.DockerImageCode.from_image_asset("./lambda"),            memory_size=1024,            timeout=Duration.seconds(30),            architecture=lambda_.Architecture.ARM_64,            environment={                "DYNAMODB_TABLE": table.table_name            }        )
        backend_function.role.add_managed_policy(            iam.ManagedPolicy.from_aws_managed_policy_name("AmazonBedrockFullAccess"))
        # Grant Lambda function permissions to access DynamoDB        table.grant_read_write_data(backend_function)
        # Create API Gateway with Cognito Authorizer        api = apigateway.RestApi(            self, "ServerlessApi",            rest_api_name="Serverless API",            description="Serverless API with Cognito authentication"        )
        authorizer = apigateway.CognitoUserPoolsAuthorizer(            self, "ApiAuthorizer",            cognito_user_pools=[user_pool]        )
        api.root.add_method(            "ANY",            apigateway.LambdaIntegration(backend_function),            authorizer=authorizer,            authorization_type=apigateway.AuthorizationType.COGNITO        )
        items = api.root.add_resource("items")        items.add_method(            "ANY",            apigateway.LambdaIntegration(backend_function),            authorizer=authorizer,            authorization_type=apigateway.AuthorizationType.COGNITO        )
        items.add_resource("{proxy+}").add_method(            "ANY",            apigateway.LambdaIntegration(backend_function),            authorizer=authorizer,            authorization_type=apigateway.AuthorizationType.COGNITO        )

requirements:accroding to this cdk project, please write a python test application which includes:1) auth congnito user pool (handle password challege is required) and get/print the id token.2) use id token to auth apigateway, test /items rest apis

import boto3import osimport requestsfrom botocore.exceptions import ClientError
def check_pro_status(custom_attrs):    """Check if user is pro and return paid status"""    try:        # Check if custom:pro exists and its value        if 'custom:pro' not in custom_attrs:            paid = False        else:            paid = custom_attrs['custom:pro'].lower() == 'yes'
        return {            'paid': paid,            'custom_attributes': custom_attrs        }
    except Exception as e:        print(f"Error checking pro status: {e}")        raise

def test_api_with_cognito_auth():    # First get the Cognito access token    # Initialize Cognito Identity Provider client    cognito_idp = boto3.client('cognito-idp')        # Replace these with your actual values    USER_POOL_ID = os.getenv('COGNITO_USER_POOL_ID')    CLIENT_ID = os.getenv('COGNITO_CLIENT_ID')    USERNAME = os.getenv('COGNITO_USERNAME')    PASSWORD = os.getenv('COGNITO_PASSWORD')        try:        # Attempt to authenticate        auth_response = cognito_idp.initiate_auth(            AuthFlow='USER_PASSWORD_AUTH',            ClientId=CLIENT_ID,            AuthParameters={                'USERNAME': USERNAME,                'PASSWORD': PASSWORD            }        )                # Check if password needs to be changed        if auth_response.get('ChallengeName') == 'NEW_PASSWORD_REQUIRED':            print("Password change required...")                        # Respond to the challenge with a new password            new_password = 'NewPassword123!'  # Replace with desired new password            challenge_response = cognito_idp.respond_to_auth_challenge(                ClientId=CLIENT_ID,                ChallengeName='NEW_PASSWORD_REQUIRED',                Session=auth_response['Session'],                ChallengeResponses={                    'USERNAME': USERNAME,                    'NEW_PASSWORD': new_password                }            )                        # Get the tokens from the challenge response            tokens = challenge_response['AuthenticationResult']            access_token = tokens['AccessToken']            print("Password changed successfully!")                    else:            # Get the tokens from the initial auth response            tokens = auth_response['AuthenticationResult']            access_token = tokens['AccessToken']            id_token = tokens['IdToken']                print(f"Access Token: {access_token}")        print(f"ID Token: {id_token}")        #create a new function to get custome attributes        response = cognito_idp.get_user(            AccessToken=access_token        )        pro = False        custom_attrs = response['UserAttributes']        if 'custom:pro' not in custom_attrs:            pro = False        else:            pro = custom_attrs['custom:pro'].lower() == 'yes'
        if pro:            print("User is pro")        else:            print("User is not pro")                # Now use the access token to call the API        api_url = os.getenv('API_ENDPOINT')        headers = {            'Content-Type': 'application/json'        }        headers = {            'Authorization': f'Bearer {access_token}',            'Content-Type': 'application/json'        }        headers = {            'Authorization': f'Bearer {id_token}',            'Content-Type': 'application/json'        }        # Test GET /items endpoint        response = requests.get(f"{api_url}/items", headers=headers)        assert response.status_code == 200        items = response.json().get('items', [])        assert isinstance(items, list)                # Test POST /items endpoint        test_item = {            "id": "test-item-1",            "name": "Test Item",            "description": "This is a test item"        }        response = requests.post(f"{api_url}/items", headers=headers, json=test_item)        assert response.status_code == 200        assert response.json()['message'] == "Item created"                # Test GET /items/<id> endpoint        response = requests.get(f"{api_url}/items/test-item-1", headers=headers)        assert response.status_code == 200        assert response.json()['id'] == "test-item-1"                # Test PUT /items/<id> endpoint        updated_item = {            "name": "Updated Test Item",            "description": "This is an updated test item"        }        response = requests.put(f"{api_url}/items/test-item-1", headers=headers, json=updated_item)        assert response.status_code == 200        assert response.json()['message'] == "Item updated"                # Test DELETE /items/<id> endpoint        response = requests.delete(f"{api_url}/items/test-item-1", headers=headers)        assert response.status_code == 200        assert response.json()['message'] == "Item deleted"        print(response.status_code)        print(response.json())
    except ClientError as e:        print(f"Authentication error: {str(e)}")        raise
if __name__ == '__main__':    test_api_with_cognito_auth()