如何强制boto3使用分块上传而非分段上传？测试类S3 API能力

当然可以通过boto3手动实现分块上传，完全可控的流程正好满足你测试S3 API处理能力和SigV4签名计算的需求。以下是具体实现步骤：

1. 初始化多部分上传 #

首先调用create_multipart_upload创建一个多部分上传任务，获取唯一的UploadId，后续所有分块操作都需要这个ID标识：

import boto3

s3_client = boto3.client('s3')
bucket_name = '你的存储桶名称'
object_key = '要上传的对象键'

# 初始化多部分上传
init_response = s3_client.create_multipart_upload(
    Bucket=bucket_name,
    Key=object_key
)
upload_id = init_response['UploadId']

2. 逐块上传文件 #

自定义分块大小，循环读取文件内容并调用upload_part上传每个分块。每个分块需要指定PartNumber（从1开始递增），boto3会自动为每个请求生成符合SigV4规范的签名：

import os

chunk_size = 10 * 1024 * 1024  # 10MB分块，可根据需求调整
file_path = '本地大文件路径'
uploaded_parts = []

with open(file_path, 'rb') as file:
    part_number = 1
    while True:
        chunk_data = file.read(chunk_size)
        if not chunk_data:
            break
        
        # 上传单个分块
        part_response = s3_client.upload_part(
            Bucket=bucket_name,
            Key=object_key,
            PartNumber=part_number,
            UploadId=upload_id,
            Body=chunk_data
        )
        
        # 记录分块的ETag和编号，后续完成上传需要
        uploaded_parts.append({
            'PartNumber': part_number,
            'ETag': part_response['ETag']
        })
        part_number += 1

3. 完成多部分上传 #

所有分块上传完成后，调用complete_multipart_upload合并分块，生成最终的S3对象：

s3_client.complete_multipart_upload(
    Bucket=bucket_name,
    Key=object_key,
    UploadId=upload_id,
    MultipartUpload={'Parts': uploaded_parts}
)

4. 异常处理与签名测试 #

• 如果上传过程中出现异常，务必调用abort_multipart_upload清理未完成的上传任务，避免存储费用浪费：

  try:
      # 上述初始化、分块上传、完成上传的逻辑
  except Exception as e:
      s3_client.abort_multipart_upload(
          Bucket=bucket_name,
          Key=object_key,
          UploadId=upload_id
      )
      raise e
  

• 关于SigV4流式签名：boto3的upload_part请求默认遵循S3的SigV4签名规范。如果需要深入测试签名逻辑，可以通过botocore的底层签名器来查看或自定义签名流程，比如：

  from botocore.signers import RequestSigner
  
  # 获取boto3使用的签名器
  signer = s3_client._request_signer
  # 可基于此构造自定义请求并生成签名，与S3的SigV4要求对比验证
  

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