对象存储 > 使用指南 > 开发指南 > AWS S3 兼容 > 兼容 SDK 示例 > 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'])

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