如何使用Python Boto3结合SQS下载S3中的未知对象？

这问题我太熟了！其实你完全不用依赖硬编码的对象名——S3推送到SQS的事件通知消息里，已经自带了触发事件的桶名和对象键。咱们只需要正确解析这些消息内容，就能动态下载对应的文件。

步骤1：搞懂SQS消息的结构 #

当你配置S3把「对象创建」事件推送到SQS时，SQS收到的消息体是JSON格式，核心信息都嵌套在Message字段里。这个字段本身又是一段JSON字符串，里面包含Records数组，每个记录就对应一个S3事件。

举个简化版的示例消息结构：

{
  "Message": "{\"Records\":[{\"s3\":{\"bucket\":{\"name\":\"my-target-bucket\"},\"object\":{\"key\":\"new-files/report-2024.csv\"}}}]}"
}

你要提取的两个关键值就在这里：

• s3.bucket.name：触发事件的S3桶名称
• s3.object.key：要下载的对象完整键（可能包含路径）

步骤2：编写解析与下载的完整代码 #

用Python的boto3库结合JSON解析，就能实现完全动态的下载逻辑。给你一个可直接参考的示例：

import boto3
import json
from urllib.parse import unquote
import os

# 初始化AWS客户端（记得配置好本地的AWS凭证）
sqs = boto3.client('sqs', region_name='your-aws-region')
s3 = boto3.client('s3', region_name='your-aws-region')

# 替换成你的SQS队列URL
QUEUE_URL = 'https://sqs.your-aws-region.amazonaws.com/123456789012/your-queue-name'
# 本地下载目录，不存在则创建
DOWNLOAD_DIR = './s3-downloads'
os.makedirs(DOWNLOAD_DIR, exist_ok=True)

def poll_and_process_sqs():
    while True:
        # 长轮询接收消息（最多10条，等待20秒减少空轮询）
        response = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20,
            AttributeNames=['All']
        )

        if 'Messages' not in response:
            print("暂无新消息，继续轮询...")
            continue

        for msg in response['Messages']:
            try:
                # 解析SQS消息体
                msg_body = json.loads(msg['Body'])
                # 解析嵌套的S3事件消息
                s3_event_data = json.loads(msg_body['Message'])
                # 提取桶名和对象键
                bucket_name = s3_event_data['Records'][0]['s3']['bucket']['name']
                object_key = s3_event_data['Records'][0]['s3']['object']['key']

                # 关键：S3对象键可能被URL编码（比如空格变成%20），必须解码
                decoded_key = unquote(object_key)
                # 本地保存路径（用对象键的最后一部分作为文件名，也可以保留完整路径）
                local_file_path = os.path.join(DOWNLOAD_DIR, os.path.basename(decoded_key))

                # 执行下载
                print(f"开始下载：{bucket_name}/{decoded_key} -> {local_file_path}")
                s3.download_file(bucket_name, object_key, local_file_path)
                print("下载完成！")

                # 处理成功后删除消息，避免重复下载
                sqs.delete_message(
                    QueueUrl=QUEUE_URL,
                    ReceiptHandle=msg['ReceiptHandle']
                )
            except Exception as e:
                print(f"处理消息失败：{str(e)}")
                # 这里可以加重试逻辑，或者把失败消息移到死信队列

if __name__ == "__main__":
    poll_and_process_sqs()

几个关键注意事项 #

• URL解码对象键：如果你的S3对象名包含空格、中文或特殊字符，S3会自动对键进行URL编码后发送到SQS，必须用unquote解码，否则download_file会找不到目标对象。
• 消息删除机制：处理完成后一定要调用delete_message，否则SQS会在「可见性超时」后重新推送这条消息，导致重复下载。
• 异常处理：添加捕获异常的逻辑可以避免单个消息处理失败导致整个轮询中断，你还可以配置死信队列来存放处理失败的消息，方便后续排查问题。
• 长轮询配置：设置WaitTimeSeconds为大于0的值（比如20秒），能减少空轮询的次数，降低API调用成本。

这样你的脚本就能自动处理任何新增到S3桶的文件，完全不用关心文件名是什么啦！

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