前言
最近遇到一个场景需要把大量的资源文件存储到 OSS 里，这里选的是腾讯的 COS 对象存储
（话说我接下来想搞的 SnapMix 项目也是需要大量存储的，我打算搭个 MinIO 把 24T 的服务器利用起来~）
为啥腾讯不搞个兼容 Amazon S3 协议的啊…… 官方的 SDK 和文档都奇奇怪怪的，感觉国内的厂商都不怎么重视文档、SDK这些，开发体验很差（特别点名微信小程序）
因为腾讯的 COS 不在 django-storages 的支持中，所以本文就没有使用这个库了，而是自己封装了一个 Storage，其实 Django 里要自定义一个 Storage 是很简单的。
OK，我在参考了一些互联网资源（以及官方文档、Github）之后，把腾讯的这个 COS 集成到 DjangoStarter 里了，不得不说 Django 这套东西还是好用，只要把 DEFAULT_FILE_STORAGE 存储后端切换到 COS ，就能实现 FileField, ImageField 这些全都自动通过 OSS 去存储和使用。
为了方便管理文件，我还用上了 django-filer 这个也算是方便，开箱即用，不过中文的 locale 有点问题，默认安装之后只能显示英文，如果需要中文得自己 fork 之后改一下（重命名 locale 目录）
本文的代码都是在 DjangoStarter 框架的基础上进行修改，在普通的 Django 项目中使用也没有问题，只是需要根据实际情况做一些修改（文件路径不同）
配置
编辑 src/config/settings/components/tencent_cos.py 文件
DEFAULT_FILE_STORAGE = "django_starter.contrib.storages.backends.TencentCOSStorage"
TENCENTCOS_STORAGE = {
# 存储桶名称，必填
"BUCKET": "",
# 存储桶文件根路径，选填，默认 '/'
"ROOT_PATH": "/",
# 上传文件时最大缓冲区大小（单位 MB），选填，默认 100
"UPLOAD_MAX_BUFFER_SIZE": 100,
# 上传文件时分块大小（单位 MB），选填，默认 10
"UPLOAD_PART_SIZE": 10,
# 上传并发上传时最大线程数，选填，默认 5
"UPLOAD_MAX_THREAD": 5,
# 腾讯云存储 Python SDK 的配置参数，详细说明请参考腾讯云官方文档。
# 注意：CONFIG中字段的大小写请与python-sdk中CosConfig的构造参数保持一致
"CONFIG": {
"Region": "ap-guangzhou",
"SecretId": "",
"SecretKey": "",
}
}
这个配置里注释都很清楚了，根据实际情况填写 bucket、id、key 等配置即可。
Storage 实现
前面有说到我把 COS 集成到 DjangoStarter 里了，所以放到了 src/django_starter/contrib 下面
安装依赖
这里需要用到腾讯提供的 Python SDK，请先安装
pdm add cos-python-sdk-v5
编写代码
编辑 src/django_starter/contrib/storages/backends/cos.py 文件。
from io import BytesIO
from shutil import copyfileobj
from tempfile import SpooledTemporaryFile
from datetime import datetime, timezone
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.core.files.storage import Storage
from django.utils.deconstruct import deconstructible
from qcloud_cos import CosConfig, CosS3Client
from qcloud_cos.cos_exception import CosServiceError
from importlib import metadata
import os.path
from django.core.files.base import File
class TencentCOSFile(File):
def __init__(self, name, storage, file=None):
super().__init__(file, name)
self.name = name
self._storage = storage
self._file = None
@property
def file(self):
if self._file is None:
self._file = SpooledTemporaryFile()
response = self._storage.client.get_object(
Bucket=self._storage.bucket,
Key=self.name,
)
raw_stream = response["Body"].get_raw_stream()
with BytesIO(raw_stream.data) as file_content:
copyfileobj(file_content, self._file)
self._file.seek(0)
return self._file
@file.setter
def file(self, value):
self._file = value
@deconstructible
class TencentCOSStorage(Storage):
"""Tencent Cloud Object Storage class for Django pluggable storage system."""
def path(self, name):
return super(TencentCOSStorage, self).path(name)
def __init__(self, bucket=None, root_path=None, config=None):
setting = getattr(settings, "TENCENTCOS_STORAGE", {})
self.bucket = bucket or setting.get("BUCKET", None)
if self.bucket is None:
raise ImproperlyConfigured("Must configure bucket.")
self.root_path = root_path or setting.get("ROOT_PATH", "/")
if not self.root_path.endswith("/"):
self.root_path += "/"
self.upload_max_buffer_size = setting.get("UPLOAD_MAX_BUFFER_SIZE", None)
self.upload_part_size = setting.get("UPLOAD_PART_SIZE", None)
self.upload_max_thread = setting.get("UPLOAD_MAX_THREAD", None)
config_kwargs = config or setting.get("CONFIG", {})
package_name = "cos-python-sdk-v5" # 替换为您要查询的包的名称
version = metadata.version(package_name)
config_kwargs["UA"] = "tencentcloud-django-plugin-cos/0.0.1;cos-python-sdk-v5/" + version
required = ["Region", "SecretId", "SecretKey"]
for key in required:
if key not in config_kwargs:
raise ImproperlyConfigured("{key} is required.".format(key=key))
config = CosConfig(**config_kwargs)
self.client = CosS3Client(config)
def _full_path(self, name):
if name == "/":
name = ""
# p = safe_join(self.root_path, name).replace("\\", "/")
# 乱起名的问题（自动在路径前加上 D:\ 之类的）终于解决了
# 腾讯哪个人才想到用 Django 内部的 safe_join 方法代替 os.path.join 的？告诉我，我绝对不打死他！！！
p = os.path.join(self.root_path, name).replace("\\", "/")
return p
def delete(self, name):
self.client.delete_object(Bucket=self.bucket, Key=self._full_path(name))
def exists(self, name):
try:
return bool(
self.client.head_object(Bucket=self.bucket, Key=self._full_path(name))
)
except CosServiceError as e:
if e.get_status_code() == 404 and e.get_error_code() == "NoSuchResource":
return False
raise
def listdir(self, path):
directories, files = [], []
full_path = self._full_path(path)
if full_path == "/":
full_path = ""
contents = []
marker = ""
while True:
# return max 1000 objects every call
response = self.client.list_objects(
Bucket=self.bucket, Prefix=full_path.lstrip("/"), Marker=marker
)
contents.extend(response["Contents"])
if response["IsTruncated"] == "false":
break
marker = response["NextMarker"]
for entry in contents:
if entry["Key"].endswith("/"):
directories.append(entry["Key"])
else:
files.append(entry["Key"])
# directories includes path itself
return directories, files
def size(self, name):
head = self.client.head_object(Bucket=self.bucket, Key=self._full_path(name))
return head["Content-Length"]
def get_modified_time(self, name):
head = self.client.head_object(Bucket=self.bucket, Key=self._full_path(name))
last_modified = head["Last-Modified"]
dt = datetime.strptime(last_modified, "%a, %d %b %Y %H:%M:%S %Z")
dt = dt.replace(tzinfo=timezone.utc)
if settings.USE_TZ:
return dt
# convert to local time
return datetime.fromtimestamp(dt.timestamp())
def get_accessed_time(self, name):
# Not implemented
return super().get_accessed_time(name)
def get_created_time(self, name):
# Not implemented
return super().get_accessed_time(name)
def url(self, name):
return self.client.get_conf().uri(
bucket=self.bucket, path=self._full_path(name)
)
def _open(self, name, mode="rb"):
tencent_cos_file = TencentCOSFile(self._full_path(name), self)
return tencent_cos_file.file
def _save(self, name, content):
upload_kwargs = {}
if self.upload_max_buffer_size is not None:
upload_kwargs["MaxBufferSize"] = self.upload_max_buffer_size
if self.upload_part_size is not None:
upload_kwargs["PartSize"] = self.upload_part_size
if self.upload_max_thread is not None:
upload_kwargs["MAXThread"] = self.upload_max_thread
self.client.upload_file_from_buffer(
self.bucket, self._full_path(name), content, **upload_kwargs
)
return os.path.relpath(name, self.root_path)
def get_available_name(self, name, max_length=None):
name = self._full_path(name)
return super().get_available_name(name, max_length)
一些絮絮叨叨：
- 这个代码是根据腾讯github上的代码修改来的，实话说写的乱七八糟，不堪入目，不过想到这也都是腾讯打工人应付工作写出来的东西，也就能理解了……
- Class 前面的
@deconstructible装饰器是 Django 内置的，用于确保在迁移时类可以被正确序列化 - 原版的代码运行起来有很多奇奇怪怪的问题，后面仔细分析了一下代码才发现，腾讯的人才好端端的
os.path.join不用，非要去用 Django 内部的 safe_join 方法，这个还是私有的，不然随便调用的… 真的逆天