在DynamoDB中用Python批量替换字符串及代码风险咨询

你的代码当前运行正常，但即使只有300条记录，还是可能遇到以下意外情况：

• 扫描结果被截断：DynamoDB的scan单次最多返回1MB数据，如果你的单条记录mytext内容很长，300条记录总大小可能超过1MB，导致response['Items']只返回部分数据，遗漏的记录不会被处理。
• 并发修改导致数据丢失：如果有其他进程同时修改这些记录，你的put_item是无条件覆盖整条记录的，会直接冲掉其他地方做的修改，没有任何冲突校验。
• 字段缺失引发崩溃：代码默认每条记录都有author、title、mytext三个字段，但如果表中存在缺失某字段的记录，会直接抛出KeyError，中断整个处理流程。
• 空值/None引发报错：如果某条记录的mytext是None或者空字符串，item["mytext"].replace(...)会直接报错，导致程序崩溃。
• 硬编码凭证的安全风险：代码里直接写死AWS密钥，一旦代码泄露，你的AWS账号权限可能被滥用，完全不符合安全规范。
• 无异常重试机制：遇到网络波动、DynamoDB临时故障时，scan或put_item会直接抛出异常中断程序，已经处理到一半的任务无法继续，也不会自动重试。

优化参考示例 #

针对上述问题可以做些调整，下面是优化后的代码片段：

import boto3
from botocore.exceptions import ClientError
import time

# 用默认凭证链获取权限，不要硬编码密钥
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('table-test')

def process_all_items():
    last_key = None
    while True:
        scan_params = {}
        if last_key:
            scan_params['ExclusiveStartKey'] = last_key
        
        # 捕获扫描异常并重试
        try:
            response = table.scan(**scan_params)
        except ClientError as e:
            print(f"扫描失败: {e.response['Error']['Message']}")
            time.sleep(1)
            continue

        for item in response['Items']:
            # 用get方法避免字段缺失或None的问题
            mytext = item.get('mytext', '')
            new_mytext = mytext.replace("?", ".")
            if mytext != new_mytext:
                # 假设author+title是复合主键，用ConditionExpression防止覆盖其他修改
                try:
                    table.put_item(
                        Item={
                            'author': item['author'],
                            'title': item['title'],
                            'mytext': new_mytext
                        },
                        ConditionExpression="author = :a AND title = :t",
                        ExpressionAttributeValues={':a': item['author'], ':t': item['title']}
                    )
                except ClientError as e:
                    print(f"更新记录 {item['author']}-{item['title']} 失败: {e.response['Error']['Message']}")
        
        last_key = response.get('LastEvaluatedKey')
        if not last_key:
            break

process_all_items()

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