PythonでDynamoDBにいろんなデータを入れてみよう〜!
このNoteは、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)
(ほぼないと思いますが。。。)
この記事が気に入ったらサポートをしてみませんか?