Python中无需循环批量检查DynamoDB用户哈希是否存在

直接用DynamoDB的BatchGetItem API就能解决你的问题，这是官方提供的批量读取接口，无需循环调用GetItem，能大幅减少网络开销。

核心思路 #

BatchGetItem允许你在单个请求中查询最多100条数据（或总数据量不超过16MB），你只需要把所有要检查的用户哈希作为分区键传入，API会返回所有存在的条目，最后对比原始列表就能找出不存在的哈希。

示例代码（Python + Boto3） #

import boto3

dynamodb = boto3.client('dynamodb')

def check_user_hashes_existence(table_name, user_hashes):
    # 构造BatchGetItem请求参数
    request_items = {
        table_name: {
            'Keys': [{'User-Hashes': {'S': hash_val}} for hash_val in user_hashes],
            # 只请求分区键，减少数据传输量
            'ProjectionExpression': 'User-Hashes'
        }
    }

    response = dynamodb.batch_get_item(RequestItems=request_items)
    
    # 提取存在的用户哈希
    existing_hashes = [item['User-Hashes']['S'] for item in response['Responses'].get(table_name, [])]
    
    # 对比原始列表，筛选出不存在的哈希
    non_existing_hashes = [hash_val for hash_val in user_hashes if hash_val not in existing_hashes]
    
    return {
        'existing': existing_hashes,
        'non_existing': non_existing_hashes
    }

# 使用示例
target_table = 'your-target-table-name'
hashes_to_verify = ['hash-001', 'hash-002', 'hash-003']
result = check_user_hashes_existence(target_table, hashes_to_verify)
print("存在的哈希:", result['existing'])
print("不存在的哈希:", result['non_existing'])

关键注意事项 #

• 如果待检查的哈希数量超过100个，需要按100个一组拆分批次处理，但即便如此，也比循环调用GetItem高效得多——批量请求大幅减少了多次网络往返的开销
• 处理UnprocessedKeys：若请求中部分键因吞吐量限制未被处理，API会返回UnprocessedKeys字段，需针对这些键发起重试请求
• 依赖Boto3默认的自动重试机制，可处理大部分临时的吞吐量限制问题

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