公共导入头：
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import sys
import logging
from django.conf import settings
from qcloud_cos.cos_exception import CosServiceError
# 正常情况日志级别使用INFO，需要定位时可以修改为DEBUG，此时SDK会打印和服务端的通信信息
# logging.basicConfig(level=logging.INFO, stream=sys.stdout)
创建桶
点击查看代码
def create_bucket(bucket, region="ap-chengdu"):
"""
创建桶
:param bucket: 桶名称 存储桶名称必须由数字、小写字母和 - 组成
:param region: 区域
:return:
"""
config = CosConfig(Region=region, SecretId=settings.TENCENT_COS_SECRETID, SecretKey=settings.TENCENT_COS_SECRETKEY)
client = CosS3Client(config)
client.create_bucket(
Bucket=bucket,
ACL="public-read" # private / public-read / public-read-write
)
cors_config = {
'CORSRule': [
{
'AllowedOrigin': '*',
'AllowedMethod': ['GET', 'PUT', 'HEAD', 'POST', 'DELETE'],
'AllowedHeader': "*",
'ExposeHeader': "*",
'MaxAgeSeconds': 500
}
]
}
client.put_bucket_cors(
Bucket=bucket,
CORSConfiguration=cors_config
)
生成一个临时凭证，并给前端返回
点击查看代码
# 生成一个临时凭证，并给前端返回
# 1. 安装一个生成临时凭证python模块 pip install -U qcloud-python-sts
# 2. 写代码
from sts.sts import Sts
def credential(bucket, region):
"""生成一个临时凭证，并给前端返回"""
config = {
# 临时密钥有效时长，单位是秒（30分钟=1800秒）
'duration_seconds': 1800,
# 固定密钥 id
'secret_id': settings.TENCENT_COS_SECRETID,
# 固定密钥 key
'secret_key': settings.TENCENT_COS_SECRETKEY,
# 换成你的 bucket
'bucket': bucket,
# 换成 bucket 所在地区
'region': region,
# 这里改成允许的路径前缀，可以根据自己网站的用户登录态判断允许上传的具体路径
# 例子： a.jpg 或者 a/* 或者 * (使用通配符*存在重大安全风险, 请谨慎评估使用)
'allow_prefix': '*',
# 密钥的权限列表。简单上传和分片需要以下的权限，其他权限列表请看 https://cloud.tencent.com/document/product/436/31923
'allow_actions': [
# 'name/cos:PostObject',
# 'name/cos:DeleteObject',
# "name/cos:UploadPart",
# "name/cos:UploadPartCopy",
# "name/cos:CompleteMultipartUpload",
# "name/cos:AbortMultipartUpload",
"*",
],
}
sts = Sts(config)
result_dict = sts.get_credential()
return result_dict
上传文件
点击查看代码
def upload_file(bucket, region, file_obj, key):
"""上传文件"""
config = CosConfig(Region=region, SecretId=settings.TENCENT_COS_SECRETID, SecretKey=settings.TENCENT_COS_SECRETKEY)
client = CosS3Client(config)
# 高级上传接口（推荐）
# 根据文件大小自动选择简单上传或分块上传，分块上传具备断点续传功能。
response = client.upload_file_from_buffer(
Bucket=bucket, # 得是已创建的bucket
Body=file_obj, # 文件对象
Key=key # 上传到桶之后的文件名（唯一）
)
# https://test-1313480500.cos.ap-chengdu.myqcloud.com/avatar.jpg
return 'https://{}.cos.{}.myqcloud.com/{}'.format(bucket, region, key)
删除文件
点击查看代码
def delete_file(bucket, region, key):
"""删除文件"""
config = CosConfig(Region=region, SecretId=settings.TENCENT_COS_SECRETID, SecretKey=settings.TENCENT_COS_SECRETKEY)
client = CosS3Client(config)
# 高级上传接口（推荐）
# 根据文件大小自动选择简单上传或分块上传，分块上传具备断点续传功能。
client.delete_object(
Bucket=bucket, # 得是已创建的bucket
Key=key # 上传到桶之后的文件名（唯一）
)
批量删除文件
点击查看代码
def delete_file_list(bucket, region, key_list):
"""批量删除文件"""
config = CosConfig(Region=region, SecretId=settings.TENCENT_COS_SECRETID, SecretKey=settings.TENCENT_COS_SECRETKEY)
client = CosS3Client(config)
"""
# 批量删除文件
objects = {
"Quiet": "true",
"Object": [
{
"Key": "file_name1"
},
{
"Key": "file_name2"
}
]
}
response = client.delete_objects(
Bucket='bucket',
Delete=objects
)
"""
objects = {
"Quiet": "true",
"Object": key_list,
}
# 高级上传接口（推荐）
# 根据文件大小自动选择简单上传或分块上传，分块上传具备断点续传功能。
client.delete_objects(
Bucket=bucket, # 得是已创建的bucket
Delete=objects # 上传到桶之后的文件名（唯一）
)
检查上传文件是否存在cos桶里
点击查看代码
def check_file(bucket, region, key):
"""检查上传文件是否存在cos桶里"""
config = CosConfig(Region=region, SecretId=settings.TENCENT_COS_SECRETID, SecretKey=settings.TENCENT_COS_SECRETKEY)
client = CosS3Client(config)
# 高级上传接口（推荐）
# 根据文件大小自动选择简单上传或分块上传，分块上传具备断点续传功能。
data = client.head_object(
Bucket=bucket, # 得是已创建的bucket
Key=key # 上传到桶之后的文件名（唯一）
)
return data
删除桶
点击查看代码
def delete_bucket(bucket, region):
"""删除桶"""
config = CosConfig(Region=region, SecretId=settings.TENCENT_COS_SECRETID, SecretKey=settings.TENCENT_COS_SECRETKEY)
client = CosS3Client(config)
try:
# 找到文件 & 删除
while True:
part_objects = client.list_objects(bucket)
# 已经删除完毕，获取不到值
contents = part_objects.get('Contents')
if not contents:
break
# 批量删除
objects = {
"Quiet": "true",
"Object": [{'Key': item["Key"]} for item in contents]
}
client.delete_objects(bucket, objects)
# 没有截断说明文件为空与上面contents效果一样
if part_objects['IsTruncated'] == "false":
break
# 找到碎片 & 删除
while True:
part_uploads = client.list_multipart_uploads(bucket)
uploads = part_uploads.get('Upload')
if not uploads:
break
for item in uploads:
client.abort_multipart_upload(bucket, item['Key'], item['UploadId'])
if part_uploads['IsTruncated'] == "false":
break
client.delete_bucket(bucket)
except CosServiceError as e:
pass