AWS SAMで構築したLambdaをDynamoDB Localに接続する
はじめに
ローカルのDynamoDBとAWS SAMで構築したLambdaを接続するのに少し苦戦したので、メモを残します。
前提
- Dockerがインストールされている
- AWS CLIがインストールされている
- LambdaのRuntimeはPython3.9
- AWS SAM CLIがインストールされている
- SAMによるプロジェクトは作成済
- venvでもanacondaでも良いが環境構築をし、boto3とulid-pyがインストールされていること
実装
テンプレート
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
my_api
Globals:
Function:
Timeout: 3
MemorySize: 128
Parameters:
SecretArn:
Type: String
Description: SecretsManagerのARN
Environment:
Type: String
Description: 環境名
Conditions:
IsProd: !Equals [ !Ref Environment, 'Prod' ]
Resources:
MyApiLayer:
Type: AWS::Serverless::LayerVersion
Properties:
LayerName:
Fn::Sub: "${Environment}-MyApiLayer"
Description: Custom layer for my_api
ContentUri: my_api_layer/
CompatibleRuntimes:
- python3.9
RetentionPolicy: Retain
MyApi:
Type: AWS::Serverless::Api
Properties:
StageName: Prod
Cors:
AllowOrigin: "'*'" # TODO: 本番環境構築時は修正
AllowMethods: "'GET,POST,PUT,DELETE'"
AllowHeaders: "'Content-Type,X-Amz-Date,Authorization,X-Api-Key,X-Amz-Security-Token'"
Auth:
ApiKeyRequired: false
MyApiRole:
Type: AWS::IAM::Role
Properties:
RoleName: MyApiRole
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
Action:
- sts:AssumeRole
Policies:
- PolicyName: BasicLambdaExecutionPolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- 'logs:CreateLogGroup'
- 'logs:CreateLogStream'
- 'logs:PutLogEvents'
Resource: '*'
- PolicyName: PostsTablePolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- dynamodb:Query
- dynamodb:Scan
- dynamodb:GetItem
- dynamodb:PutItem
- dynamodb:UpdateItem
- dynamodb:DeleteItem
Resource: !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/PostsTable"
- PolicyName: TagsTablePolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- dynamodb:Query
- dynamodb:Scan
- dynamodb:GetItem
- dynamodb:PutItem
- dynamodb:UpdateItem
- dynamodb:DeleteItem
Resource: !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${Environment}-PostsTable"
- PolicyName: ReadSecretsManagerPolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- 'secretsmanager:GetSecretValue'
Resource: !Ref SecretArn
GetPostsFunction:
Type: AWS::Serverless::Function
Properties:
Layers:
- !Ref MyApiLayer
Role: !GetAtt MyApiRole.Arn
CodeUri: functions/get_posts/
Handler: app.lambda_handler
Runtime: python3.9
FunctionName: !Sub '${Environment}-GetPostsFunction'
Architectures:
- x86_64
Events:
RootGet:
Type: Api
Properties:
RestApiId: !Ref MyApi
Path: /posts
Method: get
Environment:
Variables:
API_KEY: !Sub '{{resolve:secretsmanager:${SecretArn}:SecretString:API_KEY}}'
ENV: !Ref Environment
GetPostsFunctionLogGroup:
Type: 'AWS::Logs::LogGroup'
Properties:
LogGroupName: !Sub '/aws/lambda/${Environment}-GetPostsFunction'
RetentionInDays: 14
PostsTable:
Type: 'AWS::DynamoDB::Table'
Properties:
TableName: !Sub '${Environment}-PostsTable'
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: id
AttributeType: S
KeySchema:
- AttributeName: id
KeyType: HASH
PointInTimeRecoverySpecification:
PointInTimeRecoveryEnabled: !If [IsProd, true, false]
以上はCloudFormationで各リソースの定義です。
ローカル環境の構築とは関係ない記述も含んでいますが、念の為解説しておきます。
Parameters:
SecretArn:
Type: String
Description: SecretsManagerのARN
Environment:
Type: String
Description: 環境名
でテンプレートに渡すパラメータを定義しています。ENV
は環境名、SecretArn
はSecretsManagerのARNを設定します。これはENVでリソース名を変えたり、Lambda関数の環境変数に設定して環境ごとに読み込むリソースを変えたりし、SecretsManagerにAPIキーなどを保存しておき、lambda関数で利用するためです。
Conditions:
IsProd: !Equals [ !Ref Environment, 'Prod' ]
は環境が本番環境かどうかを定義しています。
これは本番環境のみ有効かしたい設定を制御するためです。
MyApiLayer:
Type: AWS::Serverless::LayerVersion
Properties:
LayerName:
Fn::Sub: "${Environment}-MyApiLayer"
Description: Custom layer for my_api
ContentUri: my_api_layer/
CompatibleRuntimes:
- python3.9
RetentionPolicy: Retain
はLmabda関数のレイヤーを定義しています。レイヤー内に認証やテーブルとのやりとりなどLambdaの共通処理を実装していきます。
RuntimeがpythonのLmabda関数の場合、上の定義だと
my_api_layer
|__ python
|__ my_api_layer
以下にパッケージを配置しないとLambdaが読み込めないので注意が必要です。
MyApiRole:
Type: AWS::IAM::Role
Properties:
RoleName: MyApiRole
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
Action:
- sts:AssumeRole
Policies:
- PolicyName: BasicLambdaExecutionPolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- 'logs:CreateLogGroup'
- 'logs:CreateLogStream'
- 'logs:PutLogEvents'
Resource: '*'
- PolicyName: PostsTablePolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- dynamodb:Query
- dynamodb:Scan
- dynamodb:GetItem
- dynamodb:PutItem
- dynamodb:UpdateItem
- dynamodb:DeleteItem
Resource: !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/PostsTable"
- PolicyName: TagsTablePolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- dynamodb:Query
- dynamodb:Scan
- dynamodb:GetItem
- dynamodb:PutItem
- dynamodb:UpdateItem
- dynamodb:DeleteItem
Resource: !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${Environment}-PostsTable"
- PolicyName: ReadSecretsManagerPolicy
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- 'secretsmanager:GetSecretValue'
Resource: !Ref SecretArn
はロールの定義です。DynamoDBやログ、SecretsManagerなどの権限を付与しています。
GetPostsFunction:
Type: AWS::Serverless::Function
Properties:
Layers:
- !Ref MyApiLayer
Role: !GetAtt MyApiRole.Arn
CodeUri: functions/get_posts/
Handler: app.lambda_handler
Runtime: python3.9
FunctionName: !Sub '${Environment}-GetPostsFunction'
Architectures:
- x86_64
Events:
RootGet:
Type: Api
Properties:
RestApiId: !Ref MyApi
Path: /posts
Method: get
Environment:
Variables:
API_KEY: !Sub '{{resolve:secretsmanager:${SecretArn}:SecretString:API_KEY}}'
ENV: !Ref Environment
でLambda関数を定義しています。
PostsTable:
Type: 'AWS::DynamoDB::Table'
Properties:
TableName: !Sub '${Environment}-PostsTable'
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: id
AttributeType: S
KeySchema:
- AttributeName: id
KeyType: HASH
PointInTimeRecoverySpecification:
PointInTimeRecoveryEnabled: !If [IsProd, true, false]
Postsテーブルを定義しています。ポイントタイムリカバリーの設定だけ本番環境でのみ有効にしています。
docker-composeファイル
version: '3'
services:
dynamodb-local:
image: amazon/dynamodb-local
container_name: dynamodb-local
ports:
- "8000:8000"
volumes:
- ./local-data:/home/dynamodblocal/db
command: "-jar DynamoDBLocal.jar -sharedDb -dbPath /home/dynamodblocal/db"
dynamodb-admin:
image: aaronshaf/dynamodb-admin
container_name: dynamodb-admin
environment:
- DYNAMO_ENDPOINT=http://dynamodb-local:8000
ports:
- "8001:8001"
links:
- dynamodb-local
depends_on:
- dynamodb-local
以上のdocker-compose.ymlを作成した上で
docker-compose up -d
docker network ls
を実行すると、my_api_default
(ルートディレクトリによって名前は違う)というデフォルトネットワークができます。これを後にローカルでLambda関数を起動する際に利用します。
デフォルトネットワークが嫌な方は別のdockerネットワークを定義してください。
env.json
ローカルのLambdaに環境変数を持たせるためにtemplate.ymlがあるディレクトリにenv.json
を作成します。
{
"GetPostsFunction": {
"ENV": "Local",
"API_KEY": "xxxxxxx"
}
}
エンティティ
レイヤーにエンティティを定義します。
idはソートしやすいのでulidにしていますが、uuidでも構いません。
import ulid
class PostEntity:
def __init__(self, title: str, content: str) -> None:
self.__id = str(ulid.new())
self.__title = title
self.__content = content
@classmethod
def from_dynamodb(
cls, dynamodb_item: dict
):
entity = cls.__new__(cls)
entity.__title = dynamodb_item["title"]
entity.__content = dynamodb_item["content"]
return entity
@property
def title(self):
return self.__title
@property
def content(self):
return self.__content
リポジトリ
レイヤーにリポジトリを定義します。
抽象クラスの実装は省略します。
from ..entity.post_entity import PostEntity
from .abstract_post_repository import AbstractPostRepository
import boto3
import botocore
# レイヤーのcommon.pyにテーブル名を定義(ENV + "-PostsTable")
from ..common.common import POSTS_TABLE_NAME
# レイヤーに独自例外定義
from ..exception.internal_server_exception import InternalServerException
from typing import Optional
import os
import logging
logging.getLogger()
class PostRepository(PostStudyRepository):
def __init__(self):
# ENVが"Local"またはNoneの場合はエンドポイントを指定
if os.environ.get("ENV") == "Local" or os.environ.get("ENV") == None:
self._dynamodb = boto3.resource("dynamodb", endpoint_url="http://dynamodb-local:8000")
else:
self._dynamodb = boto3.resource("dynamodb")
self._table = self._dynamodb.Table(POSTS_TABLE_NAME)
def get_posts(self):
try:
logging.info(f"get_tags")
response = self._table.scan()
except botocore.exceptions.ClientError as e:
logging.critical(f"DynamoDBからのデータ取得に失敗しました: {e}")
raise InternalServerException("サーバー内部でエラーが発生しました") from e
except Exception as e:
logging.critical(f"予期せぬエラーが発生しました: {e}")
raise InternalServerException("サーバー内部でエラーが発生しました") from e
items = response.get("Items")
if items is None:
return []
return [PostEntity.from_dynamodb(dynamodb_item=item) for item in items]
コンストラクタで、環境変数ENVが"Local"またはNoneの場合はdynamodbのエンドポイントを指定し、ローカルのDynamoDBに接続するようにしています。
サービスとapp.py
サービスで先ほど定義したリポジトリを使ってPostを取得し、dictに変換して、app.pyに返します。
app.pyでAPIのレスポンスを作成してreturnします。
実装は省略します。
injector
などのライブラリを使用することで、コンストラクタに@injectアノテーションをつけて簡単に依存関係を解決できるのでおすすめです。
テーブル作成
http://localhost:8001 にアクセスしPostsTable
を作成します。
Hash Attribute NameはStringでid
としておきます。
(後の設定は適当でいいです)
id、contentとtitleに何かを指定してレコードをいくつか作っておいてもいいかもしれません。
ローカルサーバーの起動
ビルド
sam build
起動
sam local start-api --docker-network my_api_default --env-vars env.json
--docker-networkで先ほど作成したmy_api_default
をネットワークに指定しています。また、--env-varsで先ほど定義したenv.json
を指定しています。
これでDynamoDBに接続し、http://localhost:3000/posts にPostmanでGETリクエストを送るとレコードが返ってくるはずです。
Discussion