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

# 内容

- [Python Lambda - コンテナ上でのローカルデバッグ環境構築](https://qiita.com/takmot/items/648f35c04e0717a4a6fe)
- [Python Lambda - コンテナ上でのローカルデバッグ環境構築（AWSサービス（S3）のモック化）](https://qiita.com/takmot/items/7f244baa8a56f5cd1f63)

上記過去記事の続き

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を使用

# フォルダ構成

基本は[前回記事](https://qiita.com/takmot/items/7f244baa8a56f5cd1f63#%E3%83%95%E3%82%A9%E3%83%AB%E3%83%80%E6%A7%8B%E6%88%90)から  
前回までのLambda、テストコードは省いてRDS操作用のLambda実行、テストコードを追加  

```txt
 ├─ .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 から取得してセッションを生成している

```python: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`）を定義している

```python: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テーブルに適当なレコードを追加、全データの取得を行っている

```python: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で指定

```python: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

[前回記事](https://qiita.com/takmot/items/7f244baa8a56f5cd1f63#docker-compose)からpostgresqlサービスとvolumesを追加している

```text: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 でのデバッグ

手順は[前回記事](https://qiita.com/takmot/items/648f35c04e0717a4a6fe#remote-development-%E3%81%A7%E3%81%AE%E3%83%87%E3%83%90%E3%83%83%E3%82%B0)通り  

![2023-11-11-16-23-05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/695430/b6320a35-7bbf-90b1-8c14-c87e476b4651.png)

# 参考

- [SQLAlchemyでテーブル設計とORMの操作](https://zenn.dev/shimakaze_soft/articles/6e5e47851459f5)
- [例2.Python × SQLAlchemyの場合](https://dev.classmethod.jp/articles/lambda-with-rds-tips/#toc-9)
- [Use SQLAlchemy ORM](https://github.com/aws/aws-xray-sdk-python/tree/master#use-sqlalchemy-orm)
- [docker-composeでvolumesを設定する](https://zenn.dev/ajapa/articles/1369a3c0e8085d)
- [docker上で動かしているpostgresに接続する](https://qiita.com/nakamasato/items/4155ec8acbb88af81543)
- [docker-composeでpostgresのDBが永続化できていなかったのを直した](https://qiita.com/snooow/items/04a84193fb6c7076e86f)
- [Compose の ネットワーク機能](https://docs.docker.jp/compose/networking.html)