はじめに

この記事は、「AWS AmplifyとAWS×フロントエンド Advent Calendar 2022」の12/17分です。
だいぶ後半に予定を入れたつもりでしたが、あっという間に出番が回ってきました。
いよいよ年末ですね！！

テーマに沿っていない気がしますが、この記事では「バックエンド」の話しを書きます　^^;

想定されるユースケース

想定するユースケースを説明します。

あなたは、Webシステムのバックエンド担当で、APIから後ろをメインの担当をしています。
Amplifyを使うので、構成としてはオーソドックスに、API Gateway ＋ Lambda ＋ DynamoDBを使うことにしました。

ある日のできごと（概ねフィクションです）

ユーザー　「データを変更した時に変更履歴をとっておきたい」
あなた　「変更履歴はどのように使うのでしょうか？」
ユーザー　「使い方は決まっていないが、とにかく何かの時のためにとっておきたい」
あなた　「…(*´Д｀)」

あなた　「かくかくしかじか」
開発者Ｓ　「DynamoDBの属性追加するかテーブル追加するかして保存しますか？」
あなた　「うーん。微妙だな〜...とっておけばいいってことだし...」
開発者Ｓ　「S3に保存しておきますか？」
あなた　「そうっすね。DynamoDBのStreamからLambdaに変更前データを流して、S3にputするのがいいかな？」
開発者Ｓ・あなた　「それなら今まで開発した部分に影響もないから、それで行こう！！」

このようにして、めでたく変更履歴は耐久性に優れるS3に格納することになりました。

システム構成図はこんな感じになりました

構築の流れ

前提として、Amplifyのプロジェクト内で以下のリソースは作成済みとします。

API Gateway
DynamoDB
Lambda ファンクション（DynamoDBへのCRUD操作）
変更履歴を格納する S3
なお、これらのリソースはAmplifyを用いて作成しています。

従って、ここでは、以下の内容について触れていきます。

DynamoDBのStreamから変更前イメージを受け取れるようにする
Streamをトリガーとして起動するLambdaファンクションをAmplifyで追加する
Lambdaにトリガーを追加する

1. DynamoDBのStreamから変更前イメージを受け取れるようにする

Amplifyで作ったDynamoDBは自動的にStreamがEnableになっています。（便利！）
しかし、Streamには変更後イメージだけを流すようになっています。

ここでは、マネジメントコンソールを使って、一旦StreamをDisableして、再度Enableする際に、変更前後のイメージを流すようにします。

これで、変更前のイメージをLambdaで受け取ることが出来ます。

2. Streamをトリガーとして起動するLambdaファンクションをAmplifyで追加する

このパートが唯一Amplifyを使う部分です！
amplify add function を投げるだけですが...

$ amplify add function
? Select which capability you want to add: Lambda function (serverless function)
? Provide an AWS Lambda function name: basicInfoHistoryLambda
? Choose the runtime that you want to use: Python
Only one template found - using Hello World by default.

Available advanced settings:
- Resource access permissions
- Scheduled recurring invocation
- Lambda layers configuration
- Environment variables configuration
- Secret values configuration

? Do you want to configure advanced settings? Yes
? Do you want to access other resources in this project from your Lambda function? Yes
? Select the categories you want this function to have access to. storage
? Storage has 2 resources in this project. Select the one you would like your Lambda to access s3hogehogestorage, BasicInfoDynamo
? Select the operations you want to permit on s3hogehogestorage create, read
? Select the operations you want to permit on BasicInfoDynamo read

You can access the following resource attributes as environment variables from your Lambda function
        ENV
        REGION
        STORAGE_BASICINFODYNAMO_ARN
        STORAGE_BASICINFODYNAMO_NAME
        STORAGE_BASICINFODYNAMO_STREAMARN
        STORAGE_S3HOGEHOGESTORAGE_BUCKETNAME
? Do you want to invoke this function on a recurring schedule? No
? Do you want to enable Lambda layers for this function? No
? Do you want to configure environment variables for this function? No
? Do you want to configure secret values this function can access? No
? Do you want to edit the local lambda function now? No
? Press enter to continue 
Successfully added resource basicInfoHistoryLambda locally.

これで、ローカルの amplify/backend/function/basicInfoHistoryLambdaにLambdaのひな形一式が出来ます。

Lambdaファンクションとカスタムポリシーを記載したら、amplify pushします。

Lambdaファンクション


import json
from aws_lambda_powertools import Logger
import boto3

bucket_name = "hogehoge-staging"
s3 = boto3.resource('s3')

# ロガー初期化
logger = Logger(level="INFO", service=__name__)


def handler(event, context):
  
  records = event['Records']
  for rec in records:
      logger.info(rec);
      if rec['eventName'] == 'MODIFY':
          oldImage = rec['dynamodb']['OldImage']

          id = rec['dynamodb']['Keys']['id']['S']
          approximateCreationDateTime = rec['dynamodb']['ApproximateCreationDateTime']
              
          # Dubug用のlog出力
          logger.info(id)
          logger.info(approximateCreationDateTime)
          logger.info(oldImage)
  
          # S3 に /history/<id>/<approximateCreationDateTime>.json というキーで保存
          json_key = "history/" + id + "/" + str(approximateCreationDateTime) + ".json"
          obj = s3.Object(bucket_name,json_key)

          r = obj.put(Body = json.dumps(oldImage))
  
  return {
      'statusCode': 200,
      'headers': {
          'Access-Control-Allow-Headers': '*',
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Methods': 'OPTIONS,POST,GET' },
      'body': json.dumps('Hello from your new Amplify Python lambda!')
  }

カスタムポリシー設定

LambdaがStreamを使う際にいくつか追加ポリシー設定が必要なので、custom-policies.jsonに記載します。
最初のブロックで指定するResourceには、対象のDynamoDB Streamのarnを指定します。

custom-policies.json

[
  {
    "Action": [
      "dynamodb:GetShardIterator",
      "dynamodb:DescribeStream",
      "dynamodb:GetRecords"
    ],
    "Resource": ["arn:aws:dynamodb:ap-northeast-1:<YOUR-ACCOUNT>:table/<YOUR-DynamoDB>/stream/*"]
  },
  {
    "Action": ["dynamodb:ListStreams"],
    "Resource": ["*"]
  }
]

amplify push が無事に終わったら、次に、Lambdaにトリガーを設定します。

3. Lambdaにトリガーを追加する

この操作は、マネジメントコンソールで行いまいした。

＋Add triggerボタンを押して、対象のDynamoDBを選択します。

トリガーを追加した結果は以下の様になります。

まとめ

最後までお読みいただきありがとうございます。

スキル不足もあり、マネジメントコンソールでの操作が多くなってしまいました。
これは、本来はcustomやhookを活用することで、Amplifyの操作に一本化できるのだと思います。
今後も勉強して、手順が改良出来そうでしたら、記事をアップデートしていきたいと思います。

それでは、良いAmplifyライフを！！

AmplifyでDynamoDB Streamを用いて変更履歴をS3に保存する