python - 如何使用 aws CDK 和 Python 部署层并将其附加到 aws lambda 函数

Question

如何使用 aws CDK 将层部署和附加到 aws lambda 函数？

我需要一个简单的 cdk 代码来部署一个层并将其附加到 aws lambda 函数。

Answer 1

以下 aws CDK Python 代码部署一个层并将其附加到 aws lambda 函数。

由 yl.

项目目录结构

--+
  +-app.py
  +-cdk_layers_deploy.py
  +--/functions+
               +-testLambda.py
  +--/layers+
            +-custom_func.py

app.py文件

#!/usr/bin/env python3

import sys

from aws_cdk import (core)
from cdk_layers_deploy import CdkLayersStack

app = core.App()
CdkLayersStack(app, "cdk-layers")

app.synth()

cdk_layers_deploy 文件

from aws_cdk import (
    aws_lambda as _lambda,
    core,
    aws_iam)
from aws_cdk.aws_iam import PolicyStatement
from aws_cdk.aws_lambda import LayerVersion, AssetCode


class CdkLayersStack(core.Stack):

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

        # 1) deploy lambda functions
        testLambda : _lambda.Function = CdkLayersStack.cdkResourcesCreate(self)

        # 2) attach policy to function
        projectPolicy = CdkLayersStack.createPolicy(self, testLambda)

    # -----------------------------------------------------------------------------------
    @staticmethod
    def createPolicy(this, testLambda:_lambda.Function) -> None:
        projectPolicy:PolicyStatement = PolicyStatement(
            effect=aws_iam.Effect.ALLOW,
            # resources=["*"],
            resources=[testLambda.function_arn],
            actions=[ "dynamodb:Query",
                      "dynamodb:Scan",
                      "dynamodb:GetItem",
                      "dynamodb:PutItem",
                      "dynamodb:UpdateItem",
                      "states:StartExecution",
                      "states:SendTaskSuccess",
                      "states:SendTaskFailure",
                      "cognito-idp:ListUsers",
                      "ses:SendEmail"
                    ]
        )
        return projectPolicy;
    # -----------------------------------------------------------------------------------
    @staticmethod
    def cdkResourcesCreate(self) -> None:
        lambdaFunction:_lambda.Function = _lambda.Function(self, 'testLambda',
                                                           function_name='testLambda',
                                                           handler='testLambda.lambda_handler',
                                                           runtime=_lambda.Runtime.PYTHON_3_7,
                                                           code=_lambda.Code.asset('functions'),
                                                           )
        ac = AssetCode("layers")
        layer  = LayerVersion(self, "l1", code=ac, description="test-layer", layer_version_name='Test-layer-version-name')
        lambdaFunction.add_layers(layer)

        return lambdaFunction

    # -----------------------------------------------------------------------------------

测试Lambda.py

# -------------------------------------------------
# testLambda
# -------------------------------------------------

import custom_func as cf  # this line is errored in pyCharm -> will be fixed on aws when import the layer

def lambda_handler(event, context):
    print(f"EVENT:{event}")
    ret = cf.cust_fun()
    return {
        'statusCode': 200,
        'body': ret
    }

custom_func.py - 图层函数

import datetime

def cust_fun():                           
    print("Hello from the deep layers!!")
    date_time = datetime.datetime.now().isoformat()
    print("dateTime:[%s]\n" % (date_time))

    return 1