StarBlog文字与索引
返回文章列表文章索引阅读全文
STARBLOG / 阅读手记
DjangoStarter

Django集成腾讯COS对象存储

腾讯COS集成Django真够折腾的，官方SDK和文档烂到不行。自己封装Storage虽然简单，但腾讯那帮人写代码的心思让人看不懂。Django这套东西倒是好用，`FileField`、`ImageField`自动转OSS，不过`django-filer`那中文locale问题得手动改。

曦远 (DealiAxy)约 40 分钟
目录7 个章节

前言

最近遇到一个场景需要把大量的资源文件存储到 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 目录）

PS：另外说一下，为了使用简单，我使用 django-filer 实现了在 admin 里管理静态资源，但这样流量会经过服务器，更好的做法是在前端直接上传文件到 OSS 里

本文的代码都是在 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 方法，这个还是私有的，不然随便调用的… 真的逆天

2025-7-17 更新 Storage 实现

之前的 Storage 实现不知道为啥不能用了

上传之后路径不正确，可能是 Django 新版本改了啥东西

修改后的实现是这样的，不过还有个问题，应该是 django-filer 组件自作聪明，上传的文件名中有中文（非ASCII字符）的，一律会被去掉，所以如果上传纯中文名的文件就很尴尬，文件名.jpg 变成 .jpg ，不过似乎不影响显示，我折腾了一段时间发现修改这个 storage 实现是没用的，应该得在 filer 组件那边配置或者改源码，也就不管了，凑合用吧。

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


# 关于 cos 和自定义 storage 可以参考：
# - https://www.cnblogs.com/shijieli/p/16478153.html
# - https://docs.djangoproject.com/zh-hans/4.2/ref/files/storage/
# - https://cloud.tencent.com/document/product/436/12269

@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.lstrip('/')).replace("\\", "/")

        # In __init__, self.root_path is guaranteed to end with "/"
        root = self.root_path

        # If root is just '/', treat it as an empty prefix.
        # This prevents creating absolute paths like '/path/to/file'
        # which trigger SuspiciousFileOperation.
        if root == '/':
            root = ''

        # Join the root prefix and the name, ensuring the name doesn't
        # start with a slash and uses forward slashes for URL compatibility.
        path = root + name.lstrip('/')
        return path.replace("\\", "/")

    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
        )

        # 直接返回 name，不需要额外处理
        return name

    def get_available_name(self, name, max_length=None):
        name = self._full_path(name)
        return super().get_available_name(name, max_length)

参考资料

_EOF_

Django集成腾讯COS对象存储

本文作者：
程序设计实验室
版权声明：
本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！

相关文章

讨论

共 0 条

良言一句三冬暖，恶语伤人六月寒。

加载评论中…