如何使用boto3批量检查S3存储桶中多个键是否存在？

要高效批量检查不同前缀的S3键是否存在，推荐使用S3的BatchGetObject API（对应boto3中的batch_get_object方法），一次请求最多可检查1000个对象，避免逐个调用API的低效问题。以下是具体实现：

方法一：使用BatchGetObject批量检查 #

from typing import Sequence
import boto3
from botocore.exceptions import ClientError

def keys_that_exist(bucket_name: str, keys: Sequence[str]) -> Sequence[bool]:
    s3_client = boto3.client('s3')
    exists = [False] * len(keys)
    # 按S3批量接口上限（1000个）拆分批次
    key_batches = [keys[i:i+1000] for i in range(0, len(keys), 1000)]
    
    for batch in key_batches:
        request_items = {
            bucket_name: {
                'Keys': [{'Key': key} for key in batch]
            }
        }
        
        try:
            response = s3_client.batch_get_object(RequestItems=request_items)
            # 标记存在的对象
            for obj_content in response['Responses'][bucket_name]['Contents']:
                key = obj_content['Key']
                idx = keys.index(key)
                exists[idx] = True
            
            # 处理不存在的对象（错误码为NoSuchKey）
            for error in response['Responses'][bucket_name].get('Errors', []):
                if error['Code'] != 'NoSuchKey':
                    # 非"不存在"类错误，可根据需求处理（如抛出异常或标记）
                    print(f"键 {error['Key']} 出现错误: {error['Message']}")
        except ClientError as e:
            print(f"批量请求失败: {e.response['Error']['Message']}")
    
    return exists

说明 #

• batch_get_object一次最多支持1000个对象，因此需要将键列表拆分为多个批次处理。
• 存在的对象会出现在响应的Contents字段中，直接标记为True。
• 不存在的对象会在Errors字段中返回NoSuchKey错误，此时保持初始的False即可。
• 若遇到权限不足等其他错误，可在错误分支中添加自定义处理逻辑。

方法二：异步并发检查（适合超大量键） #

如果键的数量极大，异步并发调用head_object也是高效的选择——虽然仍是单个API调用，但通过异步并行执行，效率远高于串行处理：

import asyncio
from typing import Sequence
import aioboto3
from botocore.exceptions import ClientError

async def _check_single_key(s3_client, bucket: str, key: str) -> bool:
    try:
        await s3_client.head_object(Bucket=bucket, Key=key)
        return True
    except ClientError as e:
        if e.response['Error']['Code'] == '404':
            return False
        # 其他错误抛出或自定义处理
        raise

async def keys_that_exist_async(bucket_name: str, keys: Sequence[str]) -> Sequence[bool]:
    async with aioboto3.client('s3') as s3_client:
        tasks = [_check_single_key(s3_client, bucket_name, key) for key in keys]
        return await asyncio.gather(*tasks)

# 调用示例
# results = asyncio.run(keys_that_exist_async('your-bucket', ['key1', 'key2', ...]))

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