内容

CDKで実装した Python Lambda のローカルデバッグ環境構築する
Lambdaの実行環境と合わせるため、Lambdaと同環境のコンテナを用意し、
Remote Developmentでコンテナに入りデバッグを行う

上記を前回記事で実施したので、
LocalStackでAWSサービス（S3）のモック化を行う

以下は別記事で対応する

Amazon RDSを別途PostgreSQLサーバー用コンテナで対応する

フォルダ構成

基本は前回記事通り
モック化したS3間でput/getするテスト用ファイルを test/sample-function に追加

 ├─ .devcontainer
 │     └─ devcontainer.json
 ├─ .vscode
 │     ├─ extensions.json 
 │     └─ settings.json
 │
 ├─ bin/
 │   └─ sample.ts
 │
 ├─ lib/
 │    │
 │    ├─ handlers/                           // Lambda実装
 │    │   │
 │    │   └─ sample-function/
 │    │        ├─ python/
 │    │        │     ├─ pandas/             // 外部パッケージ
 │    │        │     …
 │    │        │     └─ sample_function.py  // Lambda関数実装
 │    │        │
 │    │        └─ requirements.txt          // Lambdaに含める外部パッケージ指定
 │    │
 │    ├─ layers/                            // Layer実装
 │    │   │
 │    │   ├─ python/
 │    │   │     ├─ pytz/                    // 外部パッケージ
 │    │   │     …
 │    │   │     └─ common_layer.py          // layerに含める共通処理実装
 │    │   │
 │    │   └─ requirements.txt               // layerに含める外部パッケージ指定
 │    │
 │    └─ sample-stack.ts
 │
 ├─ test/
 │    │
 │    ├─ sample-function/
 │    │            ├─ test_data.txt         // S3間でput/getするファイル
 │    │            └─ test.py               // lib/handlers/sample-function/python/sample_function.pyのテストコード
 │    │
 │    ├─ docker-compose.yml
 │    └─ Dockerfile-lambda-python           // Python Lambda実行用のコンテナDockerfile
 …
 │
 ├─ poetry.lock
 ├─ pyproject.toml
 │

テスト対象Lambda関数コード

前回記事より、
S3からファイルをダウンロードする処理を追加している
LocalStackにアクセスするため boto3.client に endpoint_url を指定している
環境変数で ENDPOINT_URL が指定されている場合のみ適用

lib/handlers/sample-function/python/sample_function.py

import os
import boto3
from datetime import datetime
from pytz import timezone
from common_layer import print_message  # type: ignore | common layer package
import pandas as pd

test_env_value = os.getenv("TEST_ENV_VALUE")
endpoint_url = os.getenv("ENDPOINT_URL", default=None)


def handler(event, context):
    print(f"event: {event}")
    print(f"test_env_value: {test_env_value}")
    print(f"endpoint_url: {endpoint_url}")

    print_message("test message 1")

    df = pd.DataFrame([["A", 170, 60], ["B", 160, 50], ["C", 165, 58]])
    print(f"df: {df}")

    tokyo_dt = datetime.now(timezone("Asia/Tokyo"))
    
    # S3 get object
    s3_client = boto3.client('s3', endpoint_url=endpoint_url)
    response = s3_client.get_object(Bucket=event["bucket"], Key=event["key"])
    file_data = response["Body"].read().decode("utf-8")
    print(f"file_data: {file_data}")

    return {"statusCode": 200, "body": tokyo_dt.strftime("%Y-%m-%d %H:%M:%S")}

テストコード

前回記事より、
S3にBucket作成～ファイルアップロード処理を追加している
LocalStackにアクセスするため boto3.client に endpoint_url を指定している

test/sample-function/test.py

import os
import boto3
import botocore

os.environ["TEST_ENV_VALUE"] = "test env value"
os.environ["ENDPOINT_URL"] = "http://localstack:4566"   # LocalStackアクセス用のEndpoint

import sample_function

# S3 Bucket作成
def s3_init(bucket_name):
    try:
        endpoint_url = os.getenv("ENDPOINT_URL", default=None)
        s3_client = boto3.client("s3", endpoint_url=endpoint_url)
        location = {"LocationConstraint": "ap-northeast-1"}
        s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=location)
        return s3_client
    
    except botocore.exceptions.ClientError as error:
        if error.response['Error']['Code'] == 'BucketAlreadyOwnedByYou':
            print("Bucket Already Owned By You")
            return s3_client
        else:
            raise


def test_1(s3_client, test_file, bucket_name, key):
    # S3 put object
    s3_client.upload_file(test_file, bucket_name, key)
    
    event = {"bucket": bucket_name, "key": key}
    context = {}

    sample_function.handler(event, context)


test_file = "test/sample-function/test_data.txt"
bucket_name = "test-bucket-1"
key = "test_folder/test_data.txt"

s3_client = s3_init(bucket_name)
test_1(s3_client, test_file, bucket_name, key)

docker-compose

Python Lambda 実行用のコンテナについては、前回記事参照
localstackサービスを追加している

docker-compose.yml

version: "3.8"
services:
  lambda:
    container_name: lambda-python
    build:
      context: .
      dockerfile: Dockerfile-lambda-python
    volumes:
      - ../:/lambda-python
      - ~/.aws:/root/.aws
    working_dir: /lambda-python
  localstack:
    image: localstack/localstack:2.3.2
    ports:
      - "4566:4566"
    environment:
      - DEBUG=1
      - DOCKER_HOST=unix:///var/run/docker.sock

Remote Development でのデバッグ

手順は前回記事通り

Amazon RDSを別途PostgreSQLサーバー用コンテナで対応する

Python Lambda - コンテナ上でのローカルデバッグ環境構築（AWSサービス（RDS）のモック化）

Python Lambda - コンテナ上でのローカルデバッグ環境構築（AWSサービス（S3）のモック化）

内容