如何使用Python Boto3在DynamoDB事务中实现条件写入

你可以直接使用 Boto3 内置的 DynamoDB transact_write_items 接口实现需求，无需额外依赖，两个事务操作会原子性执行，要么全部成功要么全部失败。

完整代码示例 #

import boto3
from botocore.exceptions import ClientError
import time

# 初始化DynamoDB客户端，替换为你的实际AWS区域
dynamodb = boto3.client('dynamodb', region_name='us-east-1')
# 替换为你的实际表名，若两个item在不同表可分别定义
TABLE_NAME = '你的表名'

def increment_counter_when_x_expired():
    current_timestamp = int(time.time())
    try:
        response = dynamodb.transact_write_items(
            TransactItems=[
                # 第一个操作：条件检查，针对Id=x的item
                {
                    'ConditionCheck': {
                        'TableName': TABLE_NAME,
                        'Key': {
                            'Id': {'S': 'x'}
                        },
                        # 条件：x不存在，或存在但ExpiresAt小于当前时间
                        'ConditionExpression': 'attribute_not_exists(Id) OR ExpiresAt < :current_time',
                        'ExpressionAttributeValues': {
                            ':current_time': {'N': str(current_timestamp)}
                        }
                    }
                },
                # 第二个操作：更新y的counter自增1
                {
                    'Update': {
                        'TableName': TABLE_NAME,
                        'Key': {
                            'Id': {'S': 'y'}
                        },
                        # ADD操作天然支持counter不存在的场景，首次执行会自动初始化为1
                        'UpdateExpression': 'ADD counter :incr',
                        'ExpressionAttributeValues': {
                            ':incr': {'N': '1'}
                        }
                    }
                }
            ]
        )
        print("事务执行成功，counter已自增")
        return True
    except ClientError as e:
        if e.response['Error']['Code'] == 'TransactionCanceledException':
            # 事务取消通常是ConditionCheck不满足，即x存在且未过期
            print("事务取消：x存在且未过期，未执行counter自增")
            return False
        else:
            # 其他错误处理，比如权限不足、表不存在等
            print(f"执行出错：{e.response['Error']['Message']}")
            raise

# 调用示例
if __name__ == "__main__":
    increment_counter_when_x_expired()

注意事项 #

• 若你的两个item存储在不同的DynamoDB表，分别修改两个操作里的TableName参数即可
• 若使用DynamoDB的资源级接口（boto3.resource('dynamodb')），语法类似，仅参数的类型映射不需要手动指定S/N类型，接口会自动推导
• 单个DynamoDB事务最多支持25个操作，总数据量不超过4MB，你的场景完全符合限制
• 操作前请确保你的AWS账号拥有对应表的dynamodb:ConditionCheck和dynamodb:UpdateItem权限

内容的提问来源于stack exchange，提问作者troglodyte07