human: 如何处理七牛云的对象存储AWS SDK for Python

gpt: [对象存储](https://developer.qiniu.com/kodo) > 使用指南 >
[开发指南](https://developer.qiniu.com/kodo/development_guidelines/8596/development-
guidelines) > [AWS S3
兼容](https://developer.qiniu.com/kodo/development_guidelines/4086/aws-s3-compatible)
> [兼容 SDK 示例](https://developer.qiniu.com/kodo/development_guidelines/12553/S3
SDK sample) > AWS SDK for Python

#  AWS SDK for Python

最近更新时间: 2024-02-19 17:03:25

# 导入 AWS SDK for Python

确保 Python v3.7 或更新版本已经安装。

初始化 `venv` 项目

    
    
    python3 -m venv .venv
    . .venv/bin/activate
    

添加 AWS SDK for Python 包

    
    
    python3 -m pip install boto3==1.28.85
    

对于之后的每个代码示例，将代码创建在 `main.py` 后，执行

    
    
    python3 main.py
    

即可执行代码。

# 对象上传

## 获取客户端上传 URL

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    presigned_url = s3.generate_presigned_url(ClientMethod='put_object', Params={'Bucket': '<Bucket>', 'Key': '<Key>'})
    print(presigned_url)
    

这段代码将生成一个经过预先签名的客户端上传 URL，有效期为 1 小时，客户端可以在过期时间内对该 URL 发送 HTTP PUT 请求将文件上传。

以下是用 curl 上传文件的案例：

    
    
    curl -X PUT --upload-file "<path/to/file>" "<presigned url>"
    

您也可以自行指定上传凭证的有效期，例如：

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    presigned_url = s3.generate_presigned_url(ClientMethod='put_object', Params={'Bucket': '<Bucket>', 'Key': '<Key>'}, ExpiresIn=86400)
    print(presigned_url)
    

## 服务器端直传

### 单请求上传（文件）

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    with open('<path/to/upload>', 'rb') as f:
        s3.put_object(Bucket='<Bucket>', Key='<Key>', Body=f)
    

### 单请求上传（数据流）

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.put_object(Bucket='<Bucket>', Key='<Key>', Body='Hello, Qiniu S3!')
    

### 分片上传（文件）

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    create_multipart_upload_response = s3.create_multipart_upload(Bucket='<Bucket>', Key='<Key>')
    uploaded = 0
    PART_SIZE = 5*1024*1024 # 分片大小为 5 MB
    part_number = 1
    parts = []
    with open('<path/to/upload>', 'rb') as f:
        # 这里给出的案例是串行分片上传。可以自行改造成并行分片上传以进一步提升上传速度
        while True:
            part_body = f.read(PART_SIZE)
            if len(part_body) == 0:
                break
            upload_part_response = s3.upload_part(Bucket=create_multipart_upload_response['Bucket'], Key=create_multipart_upload_response['Key'],
                                                  UploadId=create_multipart_upload_response['UploadId'], PartNumber=part_number, Body=part_body)
            parts.append({'PartNumber': part_number, 'ETag': upload_part_response['ETag']})
            part_number += 1
            uploaded += len(part_body)
    complete_multipart_upload_response = s3.complete_multipart_upload(
        Bucket=create_multipart_upload_response['Bucket'], Key=create_multipart_upload_response['Key'], UploadId=create_multipart_upload_response['UploadId'], MultipartUpload={'Parts': parts})
    print('ETag:', complete_multipart_upload_response['ETag'])
    

### 上传文件

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.upload_file("<path/to/upload>", "<Bucket>", "<Key>")
    print('Done')
    

# 对象下载

## 获取客户端下载 URL

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    presigned_url = s3.generate_presigned_url(ClientMethod='get_object', Params={'Bucket': '<Bucket>', 'Key': '<Key>'})
    print(presigned_url)
    

这段代码将生成一个经过预先签名的客户端下载 URL，有效期为 1 小时，客户端可以在过期时间内对该 URL 发送 HTTP GET 请求将文件下载。

以下是用 curl 下载文件的案例：

    
    
    curl -o "<path/to/download>" "<presigned url>"
    

您也可以自行指定下载凭证的有效期，例如：

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    presigned_url = s3.generate_presigned_url(ClientMethod='get_object', Params={'Bucket': '<Bucket>', 'Key': '<Key>'}, ExpiresIn=86400)
    print(presigned_url)
    

## 服务器端直接下载

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    get_object_response = s3.get_object(Bucket='<Bucket>', Key='<Key>')
    with open('<path/to/download>', 'wb') as f:
        shutil.copyfileobj(get_object_response['Body'], f)
    print('ETag:', get_object_response['ETag'])
    

## 下载文件

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.download_file('<Bucket>', '<Key>', '<path/to/download>')
    print('Done')
    

# 对象管理

## 获取对象信息

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    head_object_response = s3.head_object(Bucket='<Bucket>', Key='<Key>')
    print('ETag:', head_object_response['ETag'])
    

## 修改对象 MimeType

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.copy_object(Bucket='<Bucket>', Key='<Key>', CopySource='/<Bucket>/<Key>', MetadataDirective='REPLACE', ContentType='<NewContentType>')
    print('Done')
    

## 修改对象存储类型

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.copy_object(Bucket='<Bucket>', Key='<Key>', CopySource='/<Bucket>/<Key>', MetadataDirective='REPLACE', StorageClass='GLACIER')
    print('Done')
    

## 复制对象副本

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.copy_object(Bucket='<ToBucket>', Key='<ToKey>', CopySource='/<FromBucket>/<FromKey>', MetadataDirective='COPY')
    print('Done')
    

## 复制对象副本（大于 5GB）

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    head_object_response = s3.head_object(
        Bucket='<FromBucket>', Key='<FromKey>')
    create_multipart_upload_response = s3.create_multipart_upload(
        Bucket='<ToBucket>', Key='<ToKey>')
    uploaded = 0
    PART_SIZE = 5*1024*1024  # 分片大小为 5 MB
    part_number = 1
    parts = []
    copied = 0
    # 这里给出的案例是串行分片上传。可以自行改造成并行分片上传以进一步提升上传速度
    while copied < head_object_response['ContentLength']:
        part_size = min(head_object_response['ContentLength'] - copied, PART_SIZE)
        upload_part_copy_response = s3.upload_part_copy(Bucket=create_multipart_upload_response['Bucket'], Key=create_multipart_upload_response['Key'],
                                                        UploadId=create_multipart_upload_response['UploadId'], PartNumber=part_number, CopySource='/<FromBucket>/<FromKey>', CopySourceRange=f'bytes={copied}-{copied+part_size}')
        parts.append({'PartNumber': part_number,
                     'ETag': upload_part_copy_response['CopyPartResult']['ETag']})
        part_number += 1
        copied += part_size
    complete_multipart_upload_response = s3.complete_multipart_upload(
        Bucket=create_multipart_upload_response['Bucket'], Key=create_multipart_upload_response['Key'], UploadId=create_multipart_upload_response['UploadId'], MultipartUpload={'Parts': parts})
    print('ETag:', complete_multipart_upload_response['ETag'])
    
    

## 删除空间中的文件

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.delete_object(Bucket='<Bucket>', Key='<Key>')
    print('Done')
    

## 获取指定前缀的文件列表

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    response = s3.list_objects_v2(Bucket='<Bucket>', Prefix='<KeyPrefix>')
    for object in response['Contents']:
        print('Key:', object['Key'])
        print('ETag:', object['ETag'])
    

## 批量删除空间中的文件

创建 `main.py`

    
    
    import boto3
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id='<QiniuAccessKey>',
                      aws_secret_access_key='<QiniuSecretKey>',
                      config=boto3.session.Config(signature_version="s3v4"))
    s3.delete_objects(Bucket='<Bucket>',
                      Delete={'Objects': [{'Key': '<Key1>'}, {'Key': '<Key2>'}, {'Key': '<Key3>'}]})
    print('Done')
    

# 临时安全凭证

创建 `main.py`

    
    
    import boto3
    
    sts = boto3.client('sts',
                       region_name='cn-east-1', # 华东-浙江区 region id
                       endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                       aws_access_key_id='<QiniuAccessKey>',
                       aws_secret_access_key='<QiniuSecretKey>',
                       config=boto3.session.Config(signature_version="s3v4"))
    get_federation_token_response = sts.get_federation_token(Name='Bob', DurationSeconds=3600, Policy='{"Version":"2012-10-17","Statement":[{"Sid":"Stmt1","Effect":"Allow","Action":["*"],"Resource":["*"]}]}')
    
    s3 = boto3.client('s3',
                      region_name='cn-east-1', # 华东-浙江区 region id
                      endpoint_url='https://s3.cn-east-1.qiniucs.com', # 华东-浙江区 endpoint
                      aws_access_key_id=get_federation_token_response['Credentials']['AccessKeyId'],
                      aws_secret_access_key=get_federation_token_response['Credentials']['SecretAccessKey'],
                      aws_session_token=get_federation_token_response['Credentials']['SessionToken'],
                      config=boto3.session.Config(signature_version="s3v4"))
    # 可以使用这些临时凭证调用 S3 服务
    list_buckets_response = s3.list_buckets()
    for bucket in list_buckets_response['Buckets']:
        print(bucket['Name'])
    

以上内容是否对您有帮助？