dynamoDBを更新するlambdaを作成

· ☕ 2

lambdaでdynamoDBを更新することってよくあるんですが、キーが違ったりすると、毎回コード書着かえたりしないといけなくて…。

めんどくさいので、汎用的に使えるのを作りました。

dyanmoDBからまず定義を取ってきて、その定義に沿ってデータ取得、更新をします。

前提

私は全dynamoDBを

{
    パーティションキー:パーティションキー
    ,"data":data
}

{
    パーティションキー:パーティションキー
    ,ソートキー:ソートキー
    ,"data":data
}

って形で持ってます。

なので、更新を掛けたい時は、dataの中の項目を更新したい事がほとんどです。

対応ケース

beforeデータ

1
2
3
4
5
6
7
{
    "userId":"00001"
    ,"data":{
        "name":"runa"
        ,"age":"26"
    }
}

投入データ

1
2
3
4
5
6
7
{
    "userId":"00001"
    ,"data":{
        "age":"27"
        ,"tel":"08012345678"
    }
}

afterデータ

1
2
3
4
5
6
7
8
{
    "userId":"00001"
    ,"data":{
        "name":"runa"
        ,"age":"27"
        ,"tel":"08012345678"
    }
}

こうしたい。

で、↑はkeyがuserIdだけど、sortKeyがあった時も、おんなじ感じで良きに計らってほしいw

コード

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import json
import boto3
from boto3.dynamodb.conditions import Key
import sys

def editData(tableName,setValue):
    dynamoDB = boto3.resource("dynamodb")
    table = dynamoDB.Table(tableName)

    #dynamoの定義を取得
    client = boto3.client('dynamodb')
    response = client.describe_table(
        TableName=tableName
    )

    keys = dict(map(lambda x:(x["KeyType"],x["AttributeName"]),response["Table"]["KeySchema"]))
    print(keys)

    
    #sortキーの有無を確認しながら、今の値を取得
    if len(response["Table"]["KeySchema"]) == 2:
        data = table.get_item(Key={keys["HASH"]:setValue[keys["HASH"]],keys["RANGE"]:setValue[keys["RANGE"]]})
    else:
        data = table.get_item(Key={keys["HASH"]:setValue[keys["HASH"]]})
        
    if "Item" in data:
        item = data["Item"]
        print(f"before:{item}")
    else:
        print("err")
        return 
    
    #更新項目の設定。
    #私の場合は、data内の項目を一つずつ更新してます。
    #適宜変更ください。
    if "data" in setValue:
        for k in setValue["data"].keys():
            item["data"][k] = setValue["data"][k]

    print(f"after:{item}")
    table.put_item(Item=item)


def main(param):
    editData(param["tableName"],param["setValue"])


def lambda_handler(event, context):

    try:
        print(json.dumps(event))
        if "body" in event:
            print(event["body"])
            return main(json.loads(event["body"]))
        elif "body-json" in event:
            print(event["body-json"])
            return main(event["body-json"])
        else:
            return main(event)
        
    except:
        import traceback
        traceback.print_exc()

lambda_handlerの中は、通常のapi-gatewayからでも、CORSのapi-gatewayからでも、lambdaの直の呼び出しでも対応できるようにしてます。

使うときは、必要な部分だけ使っていただければと思います//


るな
るな
エンジニア