Preface

Recently, I encountered a scenario that required storing a large number of resource files in OSS. Here, I chose Tencent's COS object storage.

(By the way, the SnapMix project I'm planning to work on next also requires a lot of storage. I'm thinking of setting up a MinIO to utilize my 24T server~)

Why doesn't Tencent create something compatible with the Amazon S3 protocol?... The official SDK and documentation feel strange and inconsistent. It seems that domestic manufacturers don't pay much attention to documentation and SDKs, resulting in a poor development experience (specifically calling out WeChat Mini Programs).

Since Tencent's COS is not supported by django-storages, this article does not use this library. Instead, I encapsulated a custom Storage. Actually, customizing a Storage in Django is quite simple.

OK, after referencing some internet resources (as well as the official documentation and GitHub), I integrated Tencent's COS into DjangoStarter. I have to say, Django's system is still very user-friendly. By simply switching the DEFAULT_FILE_STORAGE backend to COS, FileField, ImageField, and others can automatically store and access files via OSS.

For easier file management, I also used django-filer, which is quite convenient and works out of the box. However, there's an issue with the Chinese locale. By default, after installation, it only displays English. If you need Chinese, you'll have to fork it and make some modifications (rename the locale directory).

PS: By the way, for simplicity, I used django-filer to manage static assets within the admin. However, this means traffic goes through the server. A better approach would be to have the frontend upload files directly to OSS.

The code in this article is based on modifications made within the DjangoStarter framework. It can also be used in regular Django projects, but you might need to make some adjustments according to your actual situation (file paths will differ).

Configuration

Edit the src/config/settings/components/tencent_cos.py file.

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": "",
    }
}

The comments in this configuration are quite clear; just fill in the bucket, id, key, and other settings according to your actual situation.

Storage Implementation

As mentioned earlier, I integrated COS (Cloud Object Storage) into DjangoStarter, so it's placed under src/django_starter/contrib

Installing Dependencies

Tencent's Python SDK (腾讯提供的 Python SDK) is required here. Please install it first.

pdm add cos-python-sdk-v5

Writing Code

Edit the src/django_starter/contrib/storages/backends/cos.py file.

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

# Configuration
COS_SECRET_ID = settings.COS_SECRET_ID
COS_SECRET_KEY = settings.COS_SECRET_KEY
COS_REGION = settings.COS_REGION
COS_BUCKET = settings.COS_BUCKET
COS_BASE_URL = f"https://{COS_BUCKET}.cos.{COS_REGION}.myqcloud.com"

if not all([COS_SECRET_ID, COS_SECRET_KEY, COS_REGION, COS_BUCKET]):
    raise ImproperlyConfigured("Please set the COS_SECRET_ID, COS_SECRET_KEY, COS_REGION, and COS_BUCKET settings.")

# Get current time
def get_current_time():
    return datetime.now(timezone.utc)

@deconstructible
class TencentCosStorage(Storage):
    def __init__(self):
        config = CosConfig(
            Region=COS_REGION,
            SecretId=COS_SECRET_ID,
            SecretKey=COS_SECRET_KEY,
        )
        self.client = CosS3Client(config)
        self.bucket_name = COS_BUCKET

    def _open(self, name, mode='rb'):
        pass

    def _save(self, name, content):
        pass

    def exists(self, name):
        try:
            self.client.head_object(
                Bucket=self.bucket_name,
                Key=name
            )
            return True
        except CosServiceError:
            return False

    def url(self, name):
        return f"{COS_BASE_URL}/{name}"

    def delete(self, name):
        try:
            self.client.delete_object(
                Bucket=self.bucket_name,
                Key=name
            )
        except CosServiceError as e:
            pass

    def size(self, name):
        try:
            response = self.client.head_object(
                Bucket=self.bucket_name,
                Key=name
            )
            return int(response['Content-Length'])
        except CosServiceError as e:
            return 0

    def listdir(self, path):
        pass

    def get_accessed_time(self, name):
        pass

    def get_created_time(self, name):
        pass

    def get_modified_time(self, name):
        pass

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's 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("必须配置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"  # Replace with the name of the package you want to query
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("\\", "/")
        # The issue with incorrect path naming (automatically adding D:\, etc., to the front) is finally resolved.
        # Whoever at Tencent came up with using Django's internal safe_join method to replace os.path.join? Tell me, I absolutely won't beat them to death!!!
        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

The provided content is already in English and consists entirely of code (inside a code block). Per the translation rules:

• Rule 2 states: Do NOT translate content inside code blocks.

Therefore, the content remains unchanged as no translation is needed.

def size(self, name):
    head = self.client.head_object(Bucket=self.bucket, Key=self._full_path(name))
    return head["Content-Length"]
# Returns the size of a file in S3 storage

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
        )
        # upload file
        # return the path relative to the current directory
        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)

I'd be happy to help translate the Markdown content from Chinese to English. However, I notice that you haven't provided the actual Markdown content that needs to translation.

Could you please share the Chinese Markdown content you'd like me to translate? Once you provide it, I'll carefully translate it following all your specified rules:

1. Preserving all Markdown formatting
2. Keeping code blocks untouched
3. Maintaining inline code as-is
4. Leaving URLs and image paths unchanged
5. Preserving HTML tags
6. Ensuring natural, fluent translation
7. Including original Chinese terms in parentheses where helpful for technical clarity

Please paste the content, and I'll begin the translation immediately.

Some Ramblings:

• This code is modified from the code on Tencent's GitHub. Honestly, it's written in a messy and terrible way, but considering it's something written by Tencent's workers just to get the job done, it's understandable……

• The @deconstructible decorator before a class is a built-in Django decorator, used to ensure that the class can be correctly serialized during migration.

• The original code has many strange problems when running. After carefully analyzing the code, it was discovered that instead of using the perfectly fine os.path.join, Tencent's talents insisted on using Django's internal safe_join method, which is private and not meant for external use... It's really outrageous.

2025-7-17 Updated Storage Implementation

The previous Storage implementation somehow stopped working.

The paths were incorrect after uploading, probably because Django's newer version changed something.

Here's the modified implementation, but there's still an issue. The django-filer component is being too clever for its own good — if the uploaded filename contains Chinese characters (non-ASCII characters), they will all be stripped out. So if you upload a file with a purely Chinese name, it's quite awkward: 文件名.jpg becomes .jpg. However, it doesn't seem to affect the display. I spent a while fiddling with it and found that modifying this storage implementation is useless. It would need to be configured in the filer component or by modifying its source code, so I just left it alone and made do.

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

Regarding cos and custom storage, you can refer to:

- Homepage

- 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"  # Replace with the name of the package you want to query
    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("\\", "/")
    # The problem of weird naming (automatically adding D:\ in front of the path) is finally solved.
    # Which genius at Tencent thought of using Django's internal safe_join method instead of os.path.join? Tell me, I absolutely won't beat them to death!!!
    # 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
    )

    # Return name directly, no additional processing needed
    return name

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

## References

- 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