26 KiB
26 KiB
腾讯云 COS Python SDK 开发指南
基于腾讯云对象存储（Cloud Object Storage, COS）Python SDK 官方文档整理。
官方文档链接
目录
- 环境配置与安装
- 初始化 COS 客户端
- 存储桶操作
- 上传对象
- 下载对象
- 复制与移动对象
- 列出对象
- 判断对象是否存在
- 查询对象元数据
- 修改对象元数据
- 对象访问 URL
- 生成预签名 URL
- 恢复归档对象
- 检索对象内容
- 服务端加密
- 客户端加密
- 单链接限速
- 图片持久化处理
- 图片高级压缩
- 盲水印
- 常见问题
1. 环境配置与安装
环境要求
- Python 2.7 及 Python 3.4 及以上
安装 SDK
# 推荐: pip 安装
pip install -U cos-python-sdk-v5
# 手动安装 (从源码)
python setup.py install
# 离线安装
mkdir cos-python-sdk-packages
pip download cos-python-sdk-v5 -d cos-python-sdk-packages
tar -czvf cos-python-sdk-packages.tar.gz cos-python-sdk-packages
# 拷贝到目标机器后:
tar -xzvf cos-python-sdk-packages.tar.gz
pip install cos-python-sdk-v5 --no-index -f cos-python-sdk-packages
2. 初始化 COS 客户端
基本初始化（默认域名）
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import os
secret_id = os.environ['COS_SECRET_ID']
secret_key = os.environ['COS_SECRET_KEY']
region = 'ap-beijing' # 替换为实际 region
token = None # 永久密钥不填, 临时密钥需填入
scheme = 'https'
config = CosConfig(
Region=region,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Scheme=scheme
)
client = CosS3Client(config)
注意： 一个 region 只需生成一个
CosS3Client实例并复用，不要每次请求都创建新实例。
临时密钥初始化
config = CosConfig(
Region=region,
SecretId=tmp_secret_id,
SecretKey=tmp_secret_key,
Token=token,
Scheme=scheme
)
client = CosS3Client(config)
全球加速域名初始化
config = CosConfig(
Region=None,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Endpoint='cos.accelerate.myqcloud.com',
Scheme=scheme
)
client = CosS3Client(config)
自定义域名初始化
config = CosConfig(
Region=None,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Domain='user-define.example.com',
Scheme=scheme
)
client = CosS3Client(config)
代理初始化
proxies = {
'http': '127.0.0.1:80',
'https': '127.0.0.1:443'
}
config = CosConfig(
Region=region,
SecretId=secret_id,
SecretKey=secret_key,
Token=token,
Proxies=proxies
)
client = CosS3Client(config)
3. 存储桶操作
查询存储桶列表
response = client.list_buckets()
for bucket in response['Buckets']['Bucket']:
print(bucket['Name'], bucket['Location'], bucket['CreationDate'])
创建存储桶
response = client.create_bucket(
Bucket='examplebucket-1250000000',
ACL='private' # 可选: 'private', 'public-read', 'public-read-write'
)
检索存储桶
response = client.head_bucket(Bucket='examplebucket-1250000000')
print(response)
删除存储桶
client.delete_bucket(Bucket='examplebucket-1250000000')
判断存储桶是否存在
exists = client.bucket_exists(Bucket='examplebucket-1250000000')
print(exists) # True / False
4. 上传对象
高级接口（推荐，支持断点续传）
根据文件大小自动选择简单上传或分块上传，分块上传支持断点续传。
response = client.upload_file(
Bucket='examplebucket-1250000000',
Key='exampleobject',
LocalFilePath='local.txt',
PartSize=1, # 分块大小，单位 MB，默认 1
MAXThread=5, # 并发线程数，默认 5
EnableMD5=False
)
print(response['ETag'])
带进度回调
import sys
def upload_percentage(consumed_bytes, total_bytes):
if total_bytes:
rate = int(100 * (float(consumed_bytes) / float(total_bytes)))
print('\r{0}% '.format(rate))
sys.stdout.flush()
response = client.upload_file(
Bucket='examplebucket-1250000000',
Key='exampleobject',
LocalFilePath='local.txt',
PartSize=1,
MAXThread=5,
progress_callback=upload_percentage,
StorageClass='STANDARD',
Metadata={'x-cos-meta-key1': 'value1'}
)
断点续传（失败重试）
from qcloud_cos.cos_exception import CosClientError, CosServiceError
for i in range(10):
try:
response = client.upload_file(
Bucket='examplebucket-1250000000',
Key='exampleobject',
LocalFilePath='local.txt'
)
break
except (CosClientError, CosServiceError) as e:
print(e)
简单上传（不超过 5GB）
# 文件流上传
with open('test.txt', 'rb') as fp:
response = client.put_object(
Bucket='examplebucket-1250000000',
Body=fp,
Key='test.txt',
StorageClass='STANDARD',
ContentType='text/html; charset=utf-8'
)
print(response['ETag'])
# 字节流上传
response = client.put_object(
Bucket='examplebucket-1250000000',
Body=b'abcdefg',
Key='test.txt'
)
# 本地路径上传
response = client.put_object_from_local_file(
Bucket='examplebucket-1250000000',
LocalFilePath='local.txt',
Key='test.txt'
)
创建目录
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='path/to/create/dir/',
Body=b''
)
批量上传（文件夹上传）
from qcloud_cos.cos_threadpool import SimpleThreadPool
uploadDir = '/root/logs'
bucket = 'examplebucket-1250000000'
pool = SimpleThreadPool()
for path, dir_list, file_list in os.walk(uploadDir):
for file_name in file_list:
localFilePath = os.path.join(path, file_name)
cosObjectKey = localFilePath.strip('/')
pool.add_task(client.upload_file, bucket, cosObjectKey, localFilePath)
pool.wait_completion()
result = pool.get_result()
if not result['success_all']:
print("Not all files upload successed. you should retry")
分块上传（手动控制）
# 1. 初始化分块上传
response = client.create_multipart_upload(
Bucket='examplebucket-1250000000',
Key='exampleobject',
StorageClass='STANDARD'
)
uploadId = response['UploadId']
# 2. 上传分块（最多 10000 块）
response = client.upload_part(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Body=b'b' * 1024 * 1024,
PartNumber=1,
UploadId=uploadId
)
etag = response['ETag']
# 3. 查询已上传分块
response = client.list_parts(
Bucket='examplebucket-1250000000',
Key='exampleobject',
UploadId=uploadId
)
# 4. 完成分块上传
response = client.complete_multipart_upload(
Bucket='examplebucket-1250000000',
Key='exampleobject',
UploadId=uploadId,
MultipartUpload={
'Part': [
{'ETag': 'string', 'PartNumber': 1},
{'ETag': 'string', 'PartNumber': 2},
]
}
)
# 终止分块上传
client.abort_multipart_upload(
Bucket='examplebucket-1250000000',
Key='exampleobject',
UploadId=uploadId
)
5. 下载对象
# 下载到本地文件
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
response['Body'].get_stream_to_file('output.txt')
# 获取文件流
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
fp = response['Body'].get_raw_stream()
print(fp.read(2))
# 指定下载范围
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Range='bytes=0-10'
)
# 设置响应头部
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
ResponseContentType='text/html; charset=utf-8'
)
6. 复制与移动对象
高级复制接口（推荐）
小于 5GB 调用 copy_object，大于等于 5GB 调用分块复制。
response = client.copy(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'sourcebucket-1250000000',
'Key': 'sourceobject',
'Region': 'ap-guangzhou'
}
)
移动对象（复制 + 删除）
bucket = 'examplebucket-1250000000'
srcKey = 'src_object_key'
destKey = 'dest_object_key'
response = client.copy(
Bucket=bucket,
Key=destKey,
CopySource={
'Bucket': bucket,
'Key': srcKey,
'Region': 'ap-guangzhou'
}
)
client.delete_object(Bucket=bucket, Key=srcKey)
简单复制（不超过 5GB）
response = client.copy_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'sourcebucket-1250000000',
'Key': 'sourceobject',
'Region': 'ap-guangzhou'
}
)
7. 列出对象
列出第一页对象
response = client.list_objects(Bucket='examplebucket-1250000000')
if 'Contents' in response:
for content in response['Contents']:
print(content['Key'])
按前缀列出
response = client.list_objects(
Bucket='examplebucket-1250000000',
Prefix='folder1/'
)
分页列出全部对象
marker = ""
while True:
response = client.list_objects(
Bucket='examplebucket-1250000000',
Prefix='folder1/',
Marker=marker,
MaxKeys=10
)
if 'Contents' in response:
for content in response['Contents']:
print(content['Key'])
if response['IsTruncated'] == 'false':
break
marker = response["NextMarker"]
列出目录下的对象和子目录
response = client.list_objects(
Bucket='examplebucket-1250000000',
Prefix='folder1/',
Delimiter='/'
)
# 文件列表
if 'Contents' in response:
for content in response['Contents']:
print(content['Key'])
# 子目录
if 'CommonPrefixes' in response:
for folder in response['CommonPrefixes']:
print(folder['Prefix'])
列出对象历史版本
response = client.list_objects_versions(
Bucket='examplebucket-1250000000',
Prefix='string'
)
8. 判断对象是否存在
exists = client.object_exists(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
print(exists) # True / False
9. 查询对象元数据
response = client.head_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
print(response['Content-Length'])
print(response['ETag'])
print(response['Last-Modified'])
print(response.get('x-cos-meta-test'))
10. 修改对象元数据
通过 copy_object 并设置 CopyStatus='Replaced' 实现，源和目标是同一个对象：
response = client.copy_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'examplebucket-1250000000',
'Key': 'exampleobject',
'Region': 'ap-guangzhou'
},
CopyStatus='Replaced',
ContentType='text/plain',
Metadata={'x-cos-meta-key1': 'value1', 'x-cos-meta-key2': 'value2'}
)
修改存储类型
response = client.copy_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
CopySource={
'Bucket': 'examplebucket-1250000000',
'Key': 'exampleobject',
'Region': 'ap-guangzhou'
},
CopyStatus='Replaced',
StorageClass='STANDARD_IA'
)
11. 对象访问 URL
生成匿名下载或分发的对象访问 URL（公有读对象可直接访问，私有读需要预签名 URL）。
url = client.get_object_url(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
print(url)
12. 生成预签名 URL
上传预签名 URL
url = client.get_presigned_url(
Method='PUT',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expired=120 # 秒
)
print(url)
# 使用 URL 上传
import requests
response = requests.put(url=url, data=b'123')
限制存储类型和上传速度
url = client.get_presigned_url(
Method='PUT',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Headers={
'x-cos-storage-class': 'STANDARD_IA',
'x-cos-traffic-limit': '819200'
},
Expired=300
)
下载预签名 URL
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expired=120
)
# 指定下载文件名
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Params={
'response-content-disposition': 'attachment; filename=example.xlsx'
},
Expired=120
)
便捷下载预签名接口
url = client.get_presigned_download_url(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expired=120
)
临时密钥预签名
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
Params={'x-cos-security-token': 'string'},
Expired=120,
SignHost=False
)
域名不签入签名
当分发后使用者需修改请求域名时：
url = client.get_presigned_url(
Method='GET',
Bucket='examplebucket-1250000000',
Key='exampleobject',
SignHost=False,
Expired=120
)
13. 恢复归档对象
response = client.restore_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
RestoreRequest={
'Days': 100,
'CASJobParameters': {
'Tier': 'Standard' # 归档: Expedited/Standard/Bulk; 深度归档: Standard/Bulk
}
}
)
14. 检索对象内容
使用 SQL 从 CSV/JSON 对象中检索内容（对象需 UTF-8 编码）。
response = client.select_object_content(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Expression='Select * from COSObject',
ExpressionType='SQL',
InputSerialization={
'CompressionType': 'NONE',
'JSON': {'Type': 'LINES'}
},
OutputSerialization={
'CSV': {'RecordDelimiter': '\n'}
}
)
event_stream = response['Payload']
result = event_stream.get_select_result()
print(result)
15. 服务端加密
官方文档（对象加密）：https://cloud.tencent.com/document/product/436/63744
官方文档（存储桶加密）：https://cloud.tencent.com/document/product/436/63745
SSE-COS 加密（COS 托管密钥）
# 上传
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-cos',
Body='123',
ServerSideEncryption='AES256'
)
# 下载（自动解密，无需额外参数）
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-cos'
)
SSE-KMS 加密（KMS 托管密钥，必须 HTTPS）
import base64
from qcloud_cos.cos_comm import to_bytes
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-kms',
Body='123',
ServerSideEncryption='cos/kms',
SSEKMSKeyId='kms-key-id', # 可选，不填则用 COS 默认 CMK
SSEKMSContext=base64.standard_b64encode(to_bytes('{"test":"test"}'))
)
SSE-C 加密（用户提供密钥，必须 HTTPS）
import base64
from qcloud_cos.cos_comm import get_md5, to_bytes
ssec_secret = '00000000000000000000000000000001' # 32 字节
ssec_key = base64.standard_b64encode(to_bytes(ssec_secret))
ssec_key_md5 = get_md5(ssec_secret)
# 上传
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-c',
Body='00000',
SSECustomerAlgorithm='AES256',
SSECustomerKey=ssec_key,
SSECustomerKeyMD5=ssec_key_md5
)
# 下载（必须携带相同密钥）
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='sdk-sse-c',
SSECustomerAlgorithm='AES256',
SSECustomerKey=ssec_key,
SSECustomerKeyMD5=ssec_key_md5
)
存储桶加密
# 设置
client.put_bucket_encryption(
Bucket='examplebucket-1250000000',
ServerSideEncryptionConfiguration={
'Rule': [{'ApplySideEncryptionConfiguration': {'SSEAlgorithm': 'AES256'}}]
}
)
# 查询
response = client.get_bucket_encryption(Bucket='examplebucket-1250000000')
# 删除
client.delete_bucket_encryption(Bucket='examplebucket-1250000000')
16. 客户端加密
对称 AES256 加密
from qcloud_cos.cos_encryption_client import CosEncryptionClient
from qcloud_cos.crypto import AESProvider
# 通过密钥值初始化（aes_key_value 需 base64 编码后的 32 字节密钥）
aes_provider = AESProvider(aes_key='aes_key_value')
# 或通过密钥文件路径
# aes_provider = AESProvider(aes_key_path='aes_key_path')
client_for_aes = CosEncryptionClient(config, aes_provider)
# 上传
client_for_aes.put_object(
Bucket='examplebucket-1250000000',
Body=b'bytes',
Key='exampleobject'
)
# 下载
response = client_for_aes.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject'
)
# 断点续传上传（partsize 必须为 16 字节整数倍）
client_for_aes.upload_file(
Bucket='examplebucket-1250000000',
LocalFilePath='local.txt',
Key='exampleobject',
PartSize=10,
MAXThread=10
)
非对称 RSA 加密
from qcloud_cos.cos_encryption_client import CosEncryptionClient
from qcloud_cos.crypto import RSAProvider
rsa_key_pair = RSAProvider.get_rsa_key_pair('public_key_value', 'private_key_value')
# 或通过路径
# rsa_key_pair = RSAProvider.get_rsa_key_pair_path('public_key_path', 'private_key_path')
rsa_provider = RSAProvider(key_pair_info=rsa_key_pair)
client_for_rsa = CosEncryptionClient(config, rsa_provider)
# 上传 / 下载 / 分块上传 接口与 AES 加密客户端一致
17. 单链接限速
限速值范围：819200 - 838860800（bit/s），即 800Kb/s - 800Mb/s。
# 上传限速
with open('test.bin', 'rb') as fp:
response = client.put_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
Body=fp,
TrafficLimit='819200'
)
# 下载限速
response = client.get_object(
Bucket='examplebucket-1250000000',
Key='exampleobject',
TrafficLimit='819200'
)
response['Body'].get_stream_to_file('exampleobject')
18. 图片持久化处理
上传时处理
import json
response, data = client.ci_put_object(
Bucket='examplebucket-1250000000',
Body=open('local.jpg', 'rb'),
Key='local.jpg',
PicOperations=json.dumps({
"is_pic_info": 1,
"rules": [{
"fileid": "format.png",
"rule": "imageView2/format/png"
}]
})
)
云上数据处理
response, data = client.ci_image_process(
Bucket='examplebucket-1250000000',
Key='local.jpg',
PicOperations=json.dumps({
"is_pic_info": 1,
"rules": [{
"fileid": "format.png",
"rule": "imageView2/format/png"
}]
})
)
PicOperations rules 参数说明：
| 参数 | 描述 | 必选 |
|---|---|---|
| bucket | 目标存储桶名称，不指定则保存到当前桶 | 否 |
| fileid | 结果文件路径，以 / 开头存入指定文件夹，否则存入同目录 |
是 |
| rule | 处理参数，参见数据万象图片处理 API | 是 |
19. 图片高级压缩
官方文档：https://cloud.tencent.com/document/product/436/120479
需先在控制台开通图片高级压缩服务。
# TPG 压缩
client.ci_download_compress_image(
Bucket='examplebucket-1250000000',
Key='sample.png',
DestImagePath='sample.tpg',
CompressType='tpg'
)
# HEIF 压缩
client.ci_download_compress_image(
Bucket='examplebucket-1250000000',
Key='sample.png',
DestImagePath='sample.heif',
CompressType='heif'
)
# AVIF 压缩
client.ci_download_compress_image(
Bucket='examplebucket-1250000000',
Key='sample.png',
DestImagePath='sample.avif',
CompressType='avif'
)
支持格式：tpg、heif、avif、svg
20. 盲水印
import base64
# 添加盲水印
watermark_url = 'http://examplebucket-1250000000.cos.ap-beijing.myqcloud.com/watermark.png'
watermark_url_base64 = base64.b64encode(watermark_url.encode()).decode()
response, data = client.ci_put_object_from_local_file(
Bucket='examplebucket-1250000000',
LocalFilePath='sample.png',
Key='sample.png',
PicOperations='{"is_pic_info":1,"rules":[{"fileid":"format.png","rule":"watermark/3/type/1/image/' + watermark_url_base64 + '"}]}'
)
# 提取盲水印
sample_url = 'http://examplebucket-1250000000.cos.ap-beijing.myqcloud.com/sample.png'
sample_url_base64 = base64.b64encode(sample_url.encode()).decode()
response, data = client.ci_put_object_from_local_file(
Bucket='examplebucket-1250000000',
LocalFilePath='format.png',
Key='format.png',
PicOperations='{"is_pic_info":1,"rules":[{"fileid":"watermark.png","rule":"watermark/4/type/1/image/' + sample_url_base64 + '"}]}'
)
21. 常见问题
Q: 升级后无法执行"移动文件"操作？
COS V5 不支持移动文件接口。使用 copy_object + delete_object 实现。建议删除前校验数据 MD5 一致性。
Q: 如何获取下载文件的临时链接？
使用 get_presigned_url 或 get_presigned_download_url。使用临时密钥时需在 header 或 query_string 中加上 x-cos-security-token。
Q: 出现异常如何处理？
操作成功返回 dict 或 None。失败抛出：
CosClientError：客户端异常（网络、文件 IO、参数校验等）CosServiceError：服务端异常（文件不存在、无权限等）
Q: 上传时报 Request has expired？
签名过期，重新生成签名。若仍报错，检查机器本地时间是否为标准北京时间。
附录：存储类型
| 值 | 说明 |
|---|---|
| STANDARD | 标准存储 |
| STANDARD_IA | 低频存储 |
| ARCHIVE | 归档存储 |
附录：Region 列表
完整列表参见：https://cloud.tencent.com/document/product/436/6224
常用 Region：
| Region | 地域 |
|---|---|
| ap-beijing | 北京 |
| ap-shanghai | 上海 |
| ap-guangzhou | 广州 |
| ap-chengdu | 成都 |
| ap-hongkong | 中国香港 |
| ap-singapore | 新加坡 |
| na-siliconvalley | 硅谷 |
| eu-frankfurt | 法兰克福 |