CDK(Python)でaws_lambda_pythonのライブラリについて調べてみた

May 9, 2022

#AWS

#CDK

#Python

#LambdaPython

aws_lambda_pythonとは？

CDKの公式ドキュメントを眺めてみると、aws_lambdaの他にaws_lambda_pythonというライブラリがあることに気づきました。中身を読んでみるとaws_lambdaでは外部ライブラリを取り込む際、わりと面倒な手順が必要だったりするのをフォルダを指定するだけでライブラリ側でよしなに処理してくれるようです

まだ現時点ではEXPERIMENTALのようですが、STABLEになれば積極的に使いたいですね！

なお、aws_lambda_nodejsではSTABLEのようでした

公式ドキュメントはこちら => https://docs.aws.amazon.com/cdk/api/v2/python/aws_cdk.aws_lambda_python_alpha.html

やること

aws_lambda_pythonを使ってAPI Gateway + Lambdaの構成を実装してみました

API Gatewayについて詳細はこちら

準備

$ npm install -g aws-cdk
$ cd python
$ mkdir cdk && cd cdk
$ cdk init --language python
$ source .venv/bin/activate

requirements.txt

aws-cdk-lib==2.8.0
constructs>=10.0.0,<11.0.0
aws-cdk.aws-lambda-python-alpha

今回はCDK v2を使用します

aws-lambda-pythonはEXPERIMENTALなので-alphaをつけて別にinstallする必要があります

$ pip install -r requirements.txt

ディレクトリ構成

python/
  ├ cdk/
  │  ├ .venv
  │  ├ stacks/
  │  │    ├ __init__.py
  │  │    └ api_gateway_stack.py
  │  ├ app.py
  │  └ requirements.txt
  └ lambda/src/test_python_lambda_api/
                    ├ index.py
                    └ requirements.txt

lambda/src/test_api

/index.py

import json
import ulid

def handler(event, context):
    print(event)
    universal_id = event["pathParameters"]["universal_id"]
    ulid_value = ulid.new().str
    body = {
        'ulid': ulid.new().str,
        'universal_id': universal_id
    }
    return {
        'statusCode': 200,
        'body': json.dumps(body)
    }

/requirements.txt

ulid-py

外部ライブラリを使用したいのでulid-pyをrequirements.txtに記載します

python/cdk/stacks/api_gateway_stack.py

from aws_cdk import (
    Duration,
    Stack,
    aws_logs as logs,
    aws_lambda_python_alpha as lambda_python_,
    aws_apigateway as apigateway
)
from constructs import Construct


class ApiGatewayStack(Stack):

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

        api_python_func = lambda_python_.PythonFunction(
            self,
            "test_api_PythonLambda",
            entry="../lambda/src/test_python_lambda_api",
            runtime=lambda_.Runtime.PYTHON_3_9,
            index="index.py",
            handler="handler",
            environment={},
            timeout=Duration.seconds(
                29),
            memory_size=512,
            log_retention=logs.RetentionDays.TWO_MONTHS
        )

        api = apigateway.RestApi(self, "SampleApiGateway")
        python_univ_api = api.root.add_resource(
            "python").add_resource("{universal_id}")
        python_univ_api.add_method(
            "GET", apigateway.LambdaIntegration(api_python_func), api_key_required=True)

        api_key = apigateway.ApiKey(self, "apiKey")
        plan = api.add_usage_plan("UsagePlan")
        plan.add_api_key(api_key)
        plan.add_api_stage(stage=api.deployment_stage)

entryで指定するディレクトリに外部ライブラリを記載したrequirements.txtがあるとよしなにLambdaへライブラリをインストールしてくれます

また、PythonFunctionを使用する場合、Dockerを実行できる環境にしておく必要があるようです

/python/{universal_id}のパスでAPIGatewayを関連づけます

python/cdk/app.py

#!/usr/bin/env python3
import aws_cdk as cdk
from stacks.api_gateway_stack import ApiGatewayStack

app = cdk.App()
ApiGatewayStack(app, "ApiGatewayStack")

app.synth()

検証

$ cd cdk
$ cdk deploy

$ curl https://hogehoge.execute-api.ap-northeast-1.amazonaws.com/prod/python/yyyy1234 -H 'x-api-key:<生成したAPI Key>'

結果

{"id": "01G2ME03Y9G3RJ3SHDFPNYXVC6", "token": "yyyy1234"}

ULIDが発行されて帰ってきたのでulid-pyが想定通りインストールされていました

余談

aws_lambda.Functionを使用する場合はcodeで以下のようにすることで外部ライブラリを反映させることができます

こちらはSTABLEなのでまだ基本こっちの方法がよいかと思います

また、こちらの方法でもcdkを実行する際、Docker環境が必要です

from aws_cdk import (
    Duration,
    Stack,
    AssetHashType,
    BundlingOptions,
    aws_logs as logs,
    aws_lambda as lambda_,
    aws_apigateway as apigateway
)
from constructs import Construct


class ApiGatewayStack(Stack):

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

        api_test_func = lambda_.Function(
            self,
            "test_apiLambda",
            runtime=lambda_.Runtime.PYTHON_3_9,
            code=lambda_.Code.from_asset(
                '../lambda/src/test_api',
                asset_hash_type=AssetHashType.SOURCE,
                bundling=BundlingOptions(
                    image=lambda_.Runtime.PYTHON_3_9.bundling_image,
                    command=[
                        "bash",
                        "-c",
                        " && ".join(
                            [
                                "pip install -r requirements.txt -t /asset-output",
                                "cp -au . /asset-output",
                            ]
                        ),
                    ],
                    user="root:root"
                ),
            ),
            handler="index.handler",
            environment={},
            timeout=Duration.seconds(
                29),
            memory_size=512,
            log_retention=logs.RetentionDays.TWO_MONTHS
        )
        lambda_.Alias(self, 'test_apiLambda_alias_live',
            alias_name='Live',
            version=api_test_func.current_version
        )

        api = apigateway.RestApi(self, "SampleApiGateway")
        user_univ_token_api = api.root.add_resource(
            "user").add_resource("{universal_id}").add_resource("token")
        user_univ_token_api.add_method(
            "GET", apigateway.LambdaIntegration(api_test_func), api_key_required=True)

        api_key = apigateway.ApiKey(self, "apiKey")
        plan = api.add_usage_plan("UsagePlan")
        plan.add_api_key(api_key)
        plan.add_api_stage(stage=api.deployment_stage)

← Back to home