对象存储 > 使用指南 > 开发指南 > 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'])

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