AWS DynamoDB: ConditionalCheckFailedException

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Users')

try:
    table.put_item(
        Item={
            'user_id': 'user123',
            'email': 'test@example.com'
        },
        ConditionExpression='attribute_not_exists(user_id)'
    )
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        print('ユーザーは既に存在します')
    else:
        raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def update_with_optimistic_lock(user_id, new_data, version):
    try:
        table.update_item(
            Key={'user_id': user_id},
            UpdateExpression='SET #data = :data, #ver = :new_ver',
            ConditionExpression='#ver = :current_ver',
            ExpressionAttributeNames={
                '#data': 'data',
                '#ver': 'version'
            },
            ExpressionAttributeValues={
                ':data': new_data,
                ':new_ver': version + 1,
                ':current_ver': version
            }
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            print('競合が発生しました。再取得して再試行してください')
            raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import time
from random import uniform

def update_with_retry(user_id, update_func, max_retries=3):
    for attempt in range(max_retries):
        try:
            # 最新データを取得
            item = table.get_item(Key={'user_id': user_id})['Item']
            version = item.get('version', 0)

            # 更新を適用
            new_data = update_func(item)
            update_with_optimistic_lock(user_id, new_data, version)
            return  # 成功

        except ClientError as e:
            if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
                if attempt < max_retries - 1:
                    time.sleep(uniform(0.1, 0.5))  # ジッター付きバックオフ
                    continue
            raise

    raise Exception('最大リトライ回数を超えました')

1
2
3
4
5
6
7
8

try:
    table.delete_item(
        Key={'user_id': 'user123'},
        ConditionExpression='attribute_exists(user_id)'
    )
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        print('削除対象が存在しません')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

table.update_item(
    Key={'user_id': 'user123'},
    UpdateExpression='SET #status = :new_status',
    ConditionExpression='#status = :current_status AND #role = :admin',
    ExpressionAttributeNames={
        '#status': 'status',
        '#role': 'role'
    },
    ExpressionAttributeValues={
        ':new_status': 'active',
        ':current_status': 'pending',
        ':admin': 'admin'
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 失敗時に現在の値を取得
try:
    table.update_item(
        Key={'user_id': 'user123'},
        UpdateExpression='SET #ver = :new_ver',
        ConditionExpression='#ver = :current_ver',
        ExpressionAttributeNames={'#ver': 'version'},
        ExpressionAttributeValues={
            ':new_ver': 2,
            ':current_ver': 1
        },
        ReturnValuesOnConditionCheckFailure='ALL_OLD'
    )
except ClientError as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        current_item = e.response.get('Item', {})
        print(f"現在のバージョン: {current_item.get('version')}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

dynamodb.meta.client.transact_write_items(
    TransactItems=[
        {
            'Update': {
                'TableName': 'Users',
                'Key': {'user_id': {'S': 'user123'}},
                'UpdateExpression': 'SET balance = balance - :amount',
                'ConditionExpression': 'balance >= :amount',
                'ExpressionAttributeValues': {
                    ':amount': {'N': '100'}
                }
            }
        }
    ]
)