内容
前回記事の続き
CDKで実装した Python Lambda のローカルデバッグ環境構築する
Lambdaの実行環境と合わせるため、Lambdaと同環境のコンテナを用意し、
Remote Developmentでコンテナに入りデバッグを行う
上記を前回記事で実施したので、
LocalStackでAWSサービス（S3）のモック化を行う
以下は別記事で対応する
フォルダ構成
基本は前回記事通り
モック化した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）のモック化）