DynamoDBでドキュメント型の更新 - 闘うITエンジニアの覚え書き

DynamoDB でドキュメント型の更新では、ドキュメントの一部の項目のみを指定して更新する事はできない。
※mongoDB では $set を使う事で更新可能。

var data = { "col1" : "ABC" };
var params = {
    TableName : 'Sample',  Key: { "id" : "test01" },
    ExpressionAttributeNames: { "#col1" : "col1" },
    ExpressionAttributeValues: { ":col1" : data, },
    UpdateExpression: "SET #col1 = :col1"
};
documentClient.update(params, function(err, data) { ... }

なので、ドキュメント型の一部のみ更新したい場合でも、ドキュメント全文を指定する必要がある為、
更新前にいちど対象データのドキュメント全文を読み込んでから、更新したい箇所に変更後の値をセットして更新する必要がある。

var readParams = { ... };
documentClient.query(readParams, function(err, data) {
    data.col1 = "ABC";
    var params = {
        TableName : 'Sample',  Key: { "id" : "test01" },
        ExpressionAttributeNames: { "#col1" : "col1" },
        ExpressionAttributeValues: { ":col1" : data, },
        UpdateExpression: "SET #col1 = :col1"
    };
    documentClient.update(params, function(err, data) { ... }
});

この場合に、複数の処理で同時に処理が走る場合に、排他を考慮せずに、読み込み＆更新を行ってしまうと、
後勝ちになってしまう為、最初の更新で行われた変更が失われてしまう事がある。
※読み込みから更新までの間に、他ユーザに更新された場合。

そこで、条件付き更新を使用して更新を行う。

参考URL
http://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/Expressions.ConditionExpressions.html#Expressions.ConditionExpressions.SimpleComparisons

テーブルの作成

aws dynamodb create-table  \
    --profile developper \
    --table-name Example1 \
    --attribute-definitions \
        AttributeName=id,AttributeType=S \
    --key-schema AttributeName=id,KeyType=HASH \
    --provisioned-throughput ReadCapacityUnits=1,WriteCapacityUnits=1

var AWS = require('aws-sdk');
var documentClient = new AWS.DynamoDB.DocumentClient();

const tableName = "Example1";

const createResponse = (callback, statusCode, body) => {
    var res = {
        "statusCode": statusCode,
        "body": JSON.stringify(body)
    }
    callback(null, res);
}

const getItem = (id, callback) => {
    var params = { 
        TableName : tableName,
        KeyConditionExpression: '#id = :id',
        ExpressionAttributeNames:{ '#id': 'id'},
        ExpressionAttributeValues:{':id': id }
    };  
    documentClient.query(params, function(err, data) {
        if (err) {
            callback(err);
        } else {
             if (data.Items) {
                 data = data.Items[0] || data; 
             }
             callback(null, data);
        }   
    }); 
}

const setItemParams = (req, params, updated) => {
    var colnames = [ "col1", "col2", "col3" ];
    var delimiter = "";
    var updateExpression = "";
    for (var i in colnames) {
        var colname = colnames[i];
        if (req[colname]) {
            if (params.Item) {
                params.Item[colname] = req[colname];
            } else {
                var expressionAttributeNames = params.ExpressionAttributeNames || {};
                expressionAttributeNames["#"+colname] = colname;
                params.ExpressionAttributeNames = expressionAttributeNames;

                var expressionAttributeValues = params.ExpressionAttributeValues || {};
                expressionAttributeValues[":"+colname] = req[colname];
                params.ExpressionAttributeValues = expressionAttributeValues;

                updateExpression = updateExpression + delimiter + "#"+colname + " = " + ":" + colname;
                delimiter = ", ";
            }
        }
    }
    if (updateExpression !== "") {
        updateExpression = updateExpression + delimiter + "#updated" + " = " + ":updated";
        params.UpdateExpression = "SET " + updateExpression;
        params.ConditionExpression = " #updated = :old_updated ";
        params.ExpressionAttributeNames["#updated"] = "updated";
        // TODO: 確実ではない(ミリ秒単位で同じ時間に起動された場合)
        //              context.invokeid  や context.awsRequestId  などを利用する？
        params.ExpressionAttributeValues[":updated"] = new Date().getTime().toString();
        params.ExpressionAttributeValues[":old_updated"] = updated;
    }
    return params;
}

exports.handler = (event, context, callback) => {
    let id = false;
    if (event.pathParameters){
        id = event.pathParameters.id || false;
    }

    var req = event.body;
    if (typeof(req) === "string") {
      req = JSON.parse(req);
    }

    switch(event.httpMethod){
        case "GET":
            if(id) {
                getItem(id, function(err, data){
                    if (err) {
                        callback(err);
                        createResponse(callback, 500, { "msg": "Get Error!", "err": err, "params" : params }); 
                    } else {
                        createResponse(callback, 200, data);
                    }
                }); 
                return;
            } else {
                var params = {
                    TableName : tableName
                };
                documentClient.scan(params, function(err, data) {
                    if (err) {
                        console.log(err);
                        createResponse(callback, 500, { "msg": "List Error!", "err": err});
                    } else {
                        createResponse(callback, 200, data);
                        console.log(data);
                    }
                });
            }
            break;
        case "POST": 
            var params = { TableName : tableName , Item : { id : req.id } };
            params = setItemParams(req, params);
            documentClient.put(params, function(err, data) {
                if (err) {
                    console.log(err);
                    createResponse(callback, 500, { "msg": "Create Error!", "err": err, "req": req});
                } else {
                    createResponse(callback, 200, { "msg": "Create OK!"});
                }
            });
            break;
        case "PUT": 
            getItem(id, function(err, data){
                if (err) {
                    createResponse(callback, 500, { "msg": "Update Error!", "err": err});
                } else {
                    var sleepMs = req.sleep || 0;
                    setTimeout(function(){
                        var params = {
                            TableName : tableName,
                            Key: { "id" : id }
                        };
                        var updated = data.updated;
                        params = setItemParams(req, params, updated);
                        documentClient.update(params, function(err, data) {
                            if (err) {
                                console.log(err);
                                console.log(data);
                                createResponse(callback, 500, { "msg": "Update Error!", "err": err, "params" : params });
                            } else {
                                createResponse(callback, 200, { "msg": "Update OK!"});
                            }
                        });
                    }, sleepMs);
                }
            });
            break;
        case "DELETE": 
            var params = {
                TableName : tableName,
                Key: { "id" : parseInt(id, 10) }
            };
            documentClient.delete(params, function(err, data) {
                if (err) {
                    console.log(err);
                    console.log(data);
                    createResponse(callback, 500, { "msg": "Delete Error!", "err": err});
                } else {
                    createResponse(callback, 200, { "msg": "Delete OK!"});
                }
            });
            break;
        default:
            console.log("Error: unsupported HTTP method (" + event.httpMethod + ")");
            createResponse(callback, 501, { "msg": "Error: unsupported HTTP method (" + event.httpMethod + ")" } );
    }
}