将不同的 Lambda 函数分配给 AWS CDK 堆栈中的阶段

Question

我正在开发一个 cdk 部署脚本，我有一些工作，但我迷失了/没有运气设置不同的阶段并将不同的 lambda 应用到 api 资源。

所以我有

    // Construct lambdas - prod 
    const lambdaBacklogGet = new lambdajs.NodejsFunction(this, "name", {
      nodeModules: ['axios'],
      entry: './src/path/index.js',
      handler: 'handler',
      runtime: lambda.Runtime.NODEJS_12_X,
      timeout: cdk.Duration.seconds(20),
      role: webformRole
    });

   // Construct lambdas - dev 
    const devLambdaBacklogGet = new lambdajs.NodejsFunction(this, "name-dev", {
      nodeModules: ['axios'],
      entry: './src/path/index.js',
      handler: 'handler',
      runtime: lambda.Runtime.NODEJS_12_X,
      timeout: cdk.Duration.seconds(20),
      role: webformRole
    });

    // then I Construct API 
    const api = new apiGateway.RestApi(this, "name-api", {
      defaultCorsPreflightOptions: {
        allowOrigins: apiGateway.Cors.ALL_ORIGINS,
        allowHeaders: apiGateway.Cors.DEFAULT_HEADERS,
        allowMethods: apiGateway.Cors.ALL_METHODS,
      },
      description: "API for Something",
      deploy: true,
    });

    // Default response parameters, to allow CORS
    const corsResponseParams = {
      "method.response.header.Access-Control-Allow-Origin": true,
    };
    const corsIntegrationResponseParams = {
      "method.response.header.Access-Control-Allow-Origin": "'*'",
    };

    // REST API - /adding prod endpoint
    const backlogResourceAPI = api.root.addResource("backlog");
    const issuesBacklogGetIntegration = new apiGateway.LambdaIntegration(lambdaBacklogGet, {});
    backlogResourceAPI.addMethod("GET", issuesBacklogGetIntegration, {});

//at the end I do this
    // Then create an explicit Deployment construct
    const deployment  = new apiGateway.Deployment(this, 'my_deployment', { api });

    // And different stages
    const [devStage, testStage, prodStage] = ['dev', 'test'].map(item => 
      new apiGateway.Stage(this, `${item}_stage`, { deployment, stageName: item }));

    //api.deploymentStage = prodStage
    api.deploymentStage = devStage

我知道的最后一部分设置不正确。但基本上我想要一个 API 网关，并且我希望 prod/dev 阶段具有相同的资源/结构。但是 prod 阶段应该拉出特定的 prod lambdas，而 dev 应该调用 dev lambdas。

有没有办法通过 CDK 进行设置？

谢谢，蒂姆

Answer 1

虽然在 AWS 控制台中将 Lambda 集成添加到 API 网关时可以使用 a，Stage Variable但在编写此答案时使用 CDK 时似乎不可能。

可以Stage Variable在 S3 集成中的路径中使用 in ，但对于 Lambda 集成，我能管理的最好方法是Stage Variables从 读取event并使用 Lambda 中的逻辑，具体取决于所使用的阶段。

我的堆栈的代码（Python）：

from aws_cdk import core
from aws_cdk import aws_lambda, aws_s3
from my_cdk.api_gateway import ApiGateway


class MyCdkStack(core.Stack):

    def __init__(
        self,
        scope: core.Construct,
        construct_id: str,
        **kwargs
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        s3_bucket = aws_s3.Bucket(
            self,
            "s3_bucket",
            bucket_name="dan-dev-so-test-01"
        )

        default_lambda = aws_lambda.Function(
            self,
            "default_lambda",
            code=aws_lambda.Code.asset("./my_cdk/lambdas/"),
            handler="test_lambda.handler",
            runtime=aws_lambda.Runtime.PYTHON_3_8,
        )

        test_lambda = aws_lambda.Function(
            self,
            "test lambda",
            code=aws_lambda.Code.asset("./my_cdk/lambdas/"),
            handler="test_lambda.handler",
            runtime=aws_lambda.Runtime.PYTHON_3_8,
        )

        ApiGateway(
            self,
            "Api Gateway construct",
            default_lambda=default_lambda,
            test_lambda=test_lambda,
            s3_bucket=s3_bucket

        )

api_gateway.py 的代码

from aws_cdk import core
from aws_cdk import aws_apigateway, aws_lambda, aws_iam, aws_s3


class ApiGateway(core.Construct):

    def __init__(
        self,
        scope: core.Construct,
        construct_id: str,
        default_lambda: aws_lambda.Function,
        test_lambda: aws_lambda.Function,
        s3_bucket: aws_s3.Bucket
    ) -> None:
        super().__init__(scope, construct_id)

        api = aws_apigateway.LambdaRestApi(
            self,
            "api",
            proxy=False,
            handler=default_lambda,
            deploy_options=aws_apigateway.StageOptions(
                stage_name='v1',
                variables={
                    'stage_name': 'v1'
                }
            )
        )

        v2_deployment = aws_apigateway.Deployment(
            self,
            "deployment2",
            api=api,
            description="v2"
        )

        aws_apigateway.Stage(
            self,
            "stage 2",
            deployment=v2_deployment,
            stage_name="v2",
            variables={
                'stage_name': 'v2'
            }
        )

        test_lambda.add_permission(
            "permission",
            principal=aws_iam.ServicePrincipal('apigateway.amazonaws.com'),
            source_arn=api.arn_for_execute_api(
                stage='v2', method='GET', path='/lambdas'
            )
        )
        lambda_endpoint = api.root.add_resource("lambdas")
        s3_endpoint = api.root.add_resource("test_s3")
        lambda_endpoint.add_method(
            "GET",
            aws_apigateway.LambdaIntegration(test_lambda)
        )

        response_200 = aws_apigateway.IntegrationResponse(
            status_code="200"
        )

        method_200 =aws_apigateway.MethodResponse(
            status_code="200"
        )

        credential_role = aws_iam.Role(
            self,
            "s3_role",
            assumed_by=aws_iam.ServicePrincipal("apigateway.amazonaws.com"),
            managed_policies=[aws_iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3FullAccess")]
        )

        s3_endpoint_integration = aws_apigateway.AwsIntegration(
            service="s3",
            integration_http_method="GET",
            options=aws_apigateway.IntegrationOptions(
                credentials_role=credential_role,
                integration_responses=[response_200]
            ),
            path=f"{s3_bucket.bucket_name}/${{stageVariables.stage_name}}/file.json",


        )

        s3_endpoint.add_method(
            "GET",
            s3_endpoint_integration,
            method_responses=[method_200]
        )

Lambda 的代码：

import json


def handler(event, context):
    body = {"Error": "Not found"}
    if event['stageVariables'].get('stage_name') == 'v1':
        body = {"hi": "From V1"}
    elif event['stageVariables'].get('stage_name') == 'v2':
        body = {"hi": "From V2"}
    return {
        "statusCode": 200,
        "body": json.dumps(body)
    }

如果使用端点调用 API，/v1/lambdas它将返回：

{"hi": "From V1"}

如果使用端点调用 API，/v2/lambdas它将返回：

{"hi": "From V2"}

如果 S3 存储桶中的路径下有一个文件，v1/file.json则将使用端点返回文件内容v1/test_s3

如果 S3 存储桶中的路径下有一个文件，v2/file.json则将使用端点返回文件内容v2/test_s3