PythonでDynamoDBにいろんなデータを入れてみよう〜！

s3にデータが置かれたことをトリガーにして、起動。
s3から取得して、DynamoDBにそのまま格納する。という動きをします。

という備忘録を兼ねたメモ。

import json
import boto3
import decimal
import glob
from aws_lambda_powertools import Logger
import os

TABLE_NAME = '<DynamoDB Name>'

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(TABLE_NAME)
s3 = boto3.resource('s3')
print('Delete tmp')
for p in glob.glob('/tmp/' + '*'):
    if os.path.isfile(p):
        os.remove(p)
logger = Logger()
outbucket = os.environ['OUTBUCKET']
out_bucket = s3.Bucket(outbucket)
def lambda_handler(event, context):
    #s3からデータを取ってくる処理
    # 1. SNSトピックのeventデータ
    print('get sns')
    print(json.dumps(event, ensure_ascii=False))
    # 2. S3イベント通知のkey
    BUCKET = event['detail']['bucket']['name']
    key = event['detail']['object']['key']
    bucket = s3.Bucket(BUCKET)
    f_path = f"/tmp/{os.path.basename(key)}"
    bucket.download_file(key, f_path)
    datasize = os.path.getsize(f_path)
    if datasize < 100:
        print("reject")
        os.remove(f_path)
        return "normal end"    
    else:
        print('download fin')
        print(key)
        print(f_path)
        print(datasize,"byte")
        put_item = convert_json_dict(f_path)
        try:
           with table.batch_writer(overwrite_by_pkeys=['<partition Key>']) as batch:
               for item in put_item:
                   batch.put_item(
                       Item=item
               )
        except Exception as e:
            print(e)
def convert_json_dict(json_file_name):
    with open(json_file_name) as json_file:
        d = json.load(json_file, parse_float=decimal.Decimal)
    return d

あまりない事例かもですが、ローカルで運用してて、SSOを突破したいというのもあれば、以下のように変更すれば、、、うまくいくはずです。

my_session = boto3.Session(profile_name='<profile>')
dynamodb = my_session.resource('dynamodb')
table = dynamodb.Table(TABLE_NAME)

（ほぼないと思いますが。。。）