DynamoDBでドキュメント型項目を部分的に更新する - 闘うITエンジニアの覚え書き

例えば、以下のような内容のデータがある時、map1配下の field1 の内容だけを更新する方法について記載する。

テーブル名: TestMap、パーティションキー: id

{
  "id": "001",
  "map1": {
      "field1": {
          "value": "a1",
          "timestamp": "2018-07-01T00:00:00Z"
      },
      "field2": {
          "value": "a2",
          "timestamp": "2018-07-01T00:00:00Z"
      }
  }
}

普通に、UpdateExpression に 配下の項目を . (ドット)で区切って指定するだけで良い。
ただし、timestamp や input など、DynamoDBの 予約語を含む場合は、ExpressionAttributeNames を使用して別名を指定する必要がある。

import boto3
from boto3.dynamodb.conditions import Attr

def main():
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('TestMap')
    operation_args = {
      'Key': {'id': '001'},
      'UpdateExpression': 'set map1.field1 = :map1_field1',
      'ExpressionAttributeValues': {':map1_field1': {'value':'aa1', 'timestamp': '2018-07-23T00:00:00Z'}},
      'ReturnValues': 'UPDATED_NEW',
    }
    result = table.update_item(**operation_args)

if __name__ == '__main__':
    main()

boto3.dynamodb.conditions.Attr などを使用して 普通に ConditionExpression を指定すればOK。

以下は、map1.field1 の更新前のtimestampが指定日時より前の場合だけ更新する。
※条件を満たさない場合は、botocore.errorfactory.ConditionalCheckFailedException が発生する。

import boto3
from boto3.dynamodb.conditions import Attr

def main():
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('TestMap')
    operation_args = {
      'Key': {'id': '001'},
      'UpdateExpression': 'set map1.field1 = :map1_field1',
      'ExpressionAttributeValues': {':map1_field1': {'value':'aa1', 'timestamp': '2018-07-23T00:00:00Z'}},
      'ReturnValues': 'UPDATED_NEW',
      'ConditionExpression': Attr('map1.field1.timestamp').lt('2018-07-23T00:00:00Z')
    }
    result = table.update_item(**operation_args)

if __name__ == '__main__':
    main()

リストに要素を追加する場合は、SET句 で list_append関数を使用する。

https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/Expressions.UpdateExpressions.html#Expressions.UpdateExpressions.SET.UpdatingListElements

{
  "id": "002",
  "list1": [
    {"key": "sub1", "val": 1}
  ]
}

{
  "id": "002",
  "list1": [
    {"key": "sub1", "val": 1},
    {"key": "sub2", "val": 2}  // 追加
  ]
}

import boto3
from boto3.dynamodb.conditions import Attr

def main():
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('TestMap')
    operation_args = { 
      'Key': {'id': '002'},
      'UpdateExpression': 'SET list1 = list_append(list1, :list1_add_data)',
      'ExpressionAttributeValues': {':list1_add_data': [{'key':'sub2', 'val': 2}]},
      'ReturnValues': 'UPDATED_NEW',
    }   
    result = table.update_item(**operation_args)

def list_append(field_name, field_value):
    pass

if __name__ == '__main__':
    main()

Map配下にあるリストに要素を追加する場合も list_append で更新可能。

{
  "id": "004",
  "map1": {
    "sublist1" :[
      {"key": "sub1", "val": 1}
    ]
  ]
}

{
  "id": "004",
  "map1": {
    "sublist1" :[
      {"key": "sub1", "val": 1},
      {"key": "sub2", "val": 2}  // 追加
    ]
  ]
}

import boto3
from boto3.dynamodb.conditions import Attr

def main():
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('TestMap')
    operation_args = { 
      'Key': {'id': '003'},
      'UpdateExpression': 'SET map1.sublist1 = list_append(map1.sublist1, :list1_add_data)',
      'ExpressionAttributeValues': {':list1_add_data': [{'key':'sub2', 'val': 2}]},
      'ReturnValues': 'UPDATED_NEW',
    }   
    result = table.update_item(**operation_args)

def list_append(field_name, field_value):
    pass

if __name__ == '__main__':
    main()

以下のように、リスト配下のN番目の要素(Map)の配下にあるリストを更新する場合なども
同じ様に list_append で更新できる。

{
  "id": "004",
  "list1": [
    {
      "sublist": [
        {"key": "sub1", "val": 1}
      ]
    }
  ]
}

{
  "id": "003",
  "list1": [
    {
      "sublist": [
        {"key": "sub1", "val": 1},
        {"key": "sub2", "val": 2}  // 追加
      ]
    }
  ]
}

import boto3
from boto3.dynamodb.conditions import Attr

def main():
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('TestMap')
    operation_args = { 
      'Key': {'id': '003'},
      'UpdateExpression': 'SET list1[0].sublist = list_append(list1[0].sublist, :list1_add_data)',
      'ExpressionAttributeValues': {':list1_add_data': [{'key':'data2', 'val': 2}]},
      'ReturnValues': 'UPDATED_NEW',
    }   
    result = table.update_item(**operation_args)

def list_append(field_name, field_value):
    pass

if __name__ == '__main__':
    main()