内容
上記過去記事の続き
CDKで実装した Python Lambda のローカルデバッグ環境構築する
Lambdaの実行環境と合わせるため、Lambdaと同環境のコンテナを用意し、
Remote Developmentでコンテナに入りデバッグを行う
LocalStackでAWSサービス（S3）のモック化を行う
上記を前回までで実施したので、以下対応を行う
- PostgreSQLサーバー用コンテナでAWSサービス（RDS）のモック化を行う
LocalStackでRDSは有料サービスのため別途PostgreSQLサーバー用にコンテナを立ち上げる
RDSのシークレット（ユーザ名、パスワード）は AWS Secrets Manager に格納を想定し、
AWS Secrets Manager はLocalStackでモック化する
また、RDSの制御にはSQLAlchemyを使用
フォルダ構成
基本は前回記事から
前回までのLambda、テストコードは省いてRDS操作用のLambda実行、テストコードを追加
├─ .devcontainer
│ └─ devcontainer.json
├─ .vscode
│ ├─ extensions.json
│ └─ settings.json
│
├─ bin/
│ └─ sample.ts
│
├─ lib/
│ │
│ ├─ handlers/ // Lambda実装
│ │ │
│ │ └─ rds-sample-function/
│ │ ├─ python/
│ │ │ ├─ psycopg2/ // 外部パッケージ
│ │ │ ├─ sqlalchemy/ // 外部パッケージ
│ │ │ …
│ │ │ ├─ rds_setting.py // RDS設定ファイル
│ │ │ ├─ rds_table.py // RDSテーブル定義ファイル
│ │ │ └─ rds_sample_function.py // Lambda関数実装
│ │ │
│ │ └─ requirements.txt // Lambdaに含める外部パッケージ指定
│ …
│ │
│ └─ sample-stack.ts
│
├─ test/
│ │
│ ├─ rds-sample-function/
│ │ └─ test.py // lib/handlers/rds-sample-function/python/rds_sample_function.pyのテストコード
│ │
│ ├─ docker-compose.yml
│ └─ Dockerfile-lambda-python // Python Lambda実行用のコンテナDockerfile
…
│
├─ poetry.lock
├─ pyproject.toml
│
テスト対象Lambda関数コード
RDS設定ファイル、テーブル定義ファイル、Lambda関数実装ファイルで分割している
RDS設定ファイル
RDSのユーザ名、パスワードを Secrets Manager から取得してセッションを生成している
lib/handlers/rds-sample-function/rds_setting.py
import json
import os
import boto3 # type: ignore | lambda default install package
from sqlalchemy import create_engine # type: ignore # layer package
from sqlalchemy.ext.declarative import declarative_base # type: ignore # layer package
from sqlalchemy.orm import sessionmaker # type: ignore # layer package
endpoint_url = os.getenv("ENDPOINT_URL", default=None)
secret_manager_name = os.getenv("SECRET_MANAGER_NAME")
host = os.getenv("RDS_HOST")
database_name = os.getenv("RDS_DATABASE_NAME")
def get_secret(secret_manager_name):
client = boto3.client('secretsmanager', endpoint_url=endpoint_url)
response = client.get_secret_value(SecretId=secret_manager_name)
if "SecretString" in response:
return json.loads(response["SecretString"])
else:
raise Exception("SecretStringException")
secret = get_secret(secret_manager_name)
Engine = create_engine(
f"postgresql://{secret['username']}:{secret['password']}@{host}/{database_name}"
)
Session = sessionmaker(bind=Engine)
session = Session()
Base = declarative_base()
テーブル定義ファイル
適当なテーブル（User, DeviceData）を定義している
lib/handlers/rds-sample-function/rds_table.py
from rds_setting import Base, Engine
from sqlalchemy import Column, DateTime, Integer, String # type: ignore # layer package
class User(Base):
__tablename__ = "users"
id = Column("id", Integer, primary_key=True, autoincrement=True)
name = Column("name", String(200))
age = Column("age", Integer)
email = Column("email", String(100))
address = Column("address", String(100))
class DeviceData(Base):
__tablename__ = "device_data"
id = Column("id", Integer, primary_key=True, autoincrement=True)
date = Column("date", DateTime)
value = Column("value", Integer)
def create_all_table():
Base.metadata.create_all(bind=Engine)
Lambda関数
DeviceDataテーブルに適当なレコードを追加、全データの取得を行っている
lib/handlers/rds-sample-function/rds_sample_function.py
import datetime
import rds_setting
import rds_table
def handler(event, context):
insert_rds()
def insert_rds():
deviceData = rds_table.DeviceData()
deviceData.date = datetime.datetime.now()
deviceData.value = 12345
rds_setting.session.add(deviceData)
rds_setting.session.commit()
allData = rds_setting.session.query(rds_table.DeviceData).all()
print(f"record num: {len(allData)}")
rds_setting.session.close()
rds_setting.session.bind.dispose()
テストコード
LocalStackにアクセスするため boto3.client に endpoint_url を指定している
RDSのhost、ユーザ名、パスワード、データベース名は、docker-composeで指定
test/rds-sample-function/test.py
import json
import os
import boto3
import botocore
os.environ["ENDPOINT_URL"] = "http://localstack:4566"
os.environ["RDS_HOST"] = "postgresql"
os.environ["RDS_DATABASE_NAME"] = "postgres"
os.environ["SECRET_MANAGER_NAME"] = "/postgres/admin/secrets"
# Secrets Manager シークレット（ユーザ名、パスワード）作成
def secrets_manger_init():
endpoint_url = os.getenv("ENDPOINT_URL", default=None)
secret_manager_name = os.getenv("SECRET_MANAGER_NAME", default=None)
secret_string = json.dumps(
{"username": "postgresAdmin", "password": "postgresPassword"}
)
try:
client = boto3.client("secretsmanager", endpoint_url=endpoint_url)
client.create_secret(Name=secret_manager_name, SecretString=secret_string)
except botocore.exceptions.ClientError as error:
if error.response["Error"]["Code"] == "ResourceExistsException":
print("Resource Exists Exception")
else:
raise
secrets_manger_init()
import rds_sample_function
import rds_setting
import rds_table
# User, DeviceData テーブル作成
def rds_init():
rds_table.create_all_table()
rds_setting.session.close()
rds_setting.session.bind.dispose()
print(rds_setting.session.bind.pool.status())
# Lambda handler 実行
def test_1():
event = {}
context = {}
rds_sample_function.handler(event, context)
rds_init()
test_1()
docker-compose
前回記事からpostgresqlサービスとvolumesを追加している
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
postgresql:
container_name: postgres
image: postgres:15
volumes:
- postgresql_volume:/var/lib/postgresql/data
environment:
POSTGRES_USER: postgresAdmin
POSTGRES_PASSWORD: postgresPassword
POSTGRES_DB: postgres
ports:
- "5432:5432"
volumes:
postgresql_volume:
Remote Development でのデバッグ
手順は前回記事通り
