如何用boto3为S3存储桶中每个对象发送通知到AWS SQS

我在S3存储桶中有10个对象，希望使用boto3的list_objects_v2接口列出所有对象，并为每个对象生成一条消息发送到AWS SQS队列。我尝试了以下代码，但目前仅生成了一条通知，请问如何修改代码以实现发送10条通知？

我尝试的代码 #

import json
import boto3

region = "us-east-2"
bucket = "s3-small-files-fiap"
prefix = 'folder/'

s3_client = boto3.client('s3', region_name=region)

response = s3_client.list_objects_v2(Bucket=bucket,
                                     Prefix=prefix)

settings = {
    "bucket_name": "s3-small-files-fiap",
    "queue_name": "sqs-csv-to-json",
    "region": region,
    "account_number": <my_account_number>
}

bucket_notifications_configuration = {
    'QueueConfigurations': [{
        'Events': ['s3:ObjectCreated:*'],
        'Id': 'Notifications',
        'QueueArn':
        'arn:aws:sqs:{region}:{account_number}:{queue_name}'.format(**settings)
    }]
}

qpolicy = {
    "Version": "2012-10-17",
    "Id":
    "arn:aws:sqs:{region}:{account_number}:{queue_name}/SQSDefaultPolicy".format(
        **settings),
    "Statement": [{
        "Sid": "allow tmp bucket to notify",
        "Effect": "Allow",
        "Principal": {"AWS": "*"},
        "Action": "SQS:SendMessage",
        "Resource": "arn:aws:sqs:{region}:{account_number}:{queue_name}".format(
            **settings),
        "Condition": {
            "ArnLike": {
                "aws:SourceArn": "arn:aws:s3:*:*:{bucket_name}".format(
                    **settings)
            }
        }
    }]
}
print("Bucket notify", bucket_notifications_configuration)
print("Queue Policy", qpolicy)

queue_attrs = {"Policy": json.dumps(qpolicy), }

sqs_client = boto3.resource("sqs",
                   region_name=region).get_queue_by_name(
                       QueueName=settings["queue_name"])
sqs_client.set_attributes(Attributes=queue_attrs)
sqs_client.attributes

s3_client.put_bucket_notification_configuration(
    Bucket=bucket,
    NotificationConfiguration=bucket_notifications_configuration)

当前输出 #

Bucket notify {'QueueConfigurations': [{'Events': ['s3:ObjectCreated:*'], 'Id': 'Notifications', 'QueueArn': 'arn:aws:sqs:us-east-2:<my_account_number>:sqs-csv-to-json'}]}
Queue Policy {'Version': '2012-10-17', 'Id': 'arn:aws:sqs:us-east-2:<my_account_number>:sqs-csv-to-json/SQSDefaultPolicy', 'Statement': [{'Sid': 'allow tmp bucket to notify', 'Effect': 'Allow', 'Principal': {'AWS': '*'}, 'Action': 'SQS:SendMessage', 'Resource': 'arn:aws:sqs:us-east-2:<my_account_number>:sqs-csv-to-json', 'Condition': {'ArnLike': {'aws:SourceArn': 'arn:aws:s3:*:*:s3-small-files-fiap'}}}]}

解决方案 #

你当前的代码是配置S3桶的事件通知规则，仅当有新对象创建时才会自动发送通知到SQS，不会处理桶中已存在的对象。要为现有10个对象各发送一条SQS消息，需要添加遍历对象并主动发送消息的逻辑，修改后的代码如下：

import json
import boto3

region = "us-east-2"
bucket = "s3-small-files-fiap"
prefix = 'folder/'

# 初始化S3和SQS客户端
s3_client = boto3.client('s3', region_name=region)
sqs_resource = boto3.resource("sqs", region_name=region)

# 列出S3桶中的所有对象
response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)

# 获取目标SQS队列
settings = {
    "bucket_name": "s3-small-files-fiap",
    "queue_name": "sqs-csv-to-json",
    "region": region,
    "account_number": <my_account_number>
}
sqs_queue = sqs_resource.get_queue_by_name(QueueName=settings["queue_name"])

# 遍历每个对象，发送消息到SQS
if 'Contents' in response:
    for obj in response['Contents']:
        # 构造消息内容，可根据需求调整字段
        message_body = json.dumps({
            "bucket": bucket,
            "object_key": obj['Key'],
            "size": obj['Size'],
            "last_modified": obj['LastModified'].isoformat()
        })
        # 发送消息到SQS
        send_response = sqs_queue.send_message(MessageBody=message_body)
        print(f"已发送对象 {obj['Key']} 的通知，消息ID: {send_response['MessageId']}")
else:
    print("指定前缀下没有找到对象")

关键修改说明 #

1. 遍历现有对象：从list_objects_v2的响应中取出Contents列表，逐个处理每个S3对象
2. 构造消息体：为每个对象生成包含关键信息（如桶名、对象键、大小等）的JSON格式消息
3. 主动发送消息：调用SQS队列的send_message方法，为每个对象单独发送一条消息

注意事项 #

• 如果桶中对象数量超过1000个，list_objects_v2会返回分页结果，需要通过NextContinuationToken循环获取所有对象
• 你原代码中的队列权限配置已满足S3发送消息的需求，若仅处理现有对象，这部分可保留或按需调整

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