Django+Celery文件上传异常：文件存存储未入库求助

我正在尝试使用Celery任务处理文件上传，但所有任务完成后，查询/files接口时发现文件已存储在存储系统中，但未保存到数据库；而作为第二个任务生成的缩略图（速度更快）既存在于存储系统又已保存到数据库中。请问可能是什么问题？

相关代码 #

views.py #

@extend_schema(
        description="Upload file",
        request=FileSerializer,
        responses={201: FileSerializer},
    )
    def post(self, request, *args, **kwargs):

        file = request.data["file"]

        total_sum = File.objects.filter(owner=request.user).aggregate(models.Sum("size"))
        total_sum = total_sum["size__sum"] if total_sum["size__sum"] else 0

        if total_sum + file.size > request.user.storage_limit:
            return Response(
                {"error": "Storage limit exceeded"}, status=status.HTTP_400_BAD_REQUEST
            )

        request.data["size"] = file.size

        serializer = FileSerializer(data=request.data)

        if serializer.is_valid():
            serializer.save(owner=request.user, size=file.size)
            FileService.upload_file(serializer.data["id"], file.read(), file.content_type)

            return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

services.py #

import base64
import mimetypes
from common.service import Service
from storage.tasks import UploadHandler, ThumbnailHandler


class FileService(Service):  # Service是空类，仅定义为class Service: ...
    handlers = [UploadHandler, ThumbnailHandler]
    handler = handlers[0]()
    for next_handler in handlers[1:]:
        handler.set_next(next_handler())

    @staticmethod
    def get_file_extension(mimetype: str):
        return mimetypes.guess_extension(mimetype)

    @staticmethod
    def upload_file(file_id: str, file: bytes, mimetype: str):
        file_bytes = base64.b64encode(file)
        extension = FileService.get_file_extension(mimetype)

        FileService.handler.handle((file_id, extension, file_bytes))

tasks.py #

import base64

from celery import shared_task
from django.conf import settings
from django.core.files.base import ContentFile
from services.grpc_client import Client
from common.chain import AbstractHandler
from storage.models import File


def next_status(file: File):
    file.status += 1
    file.save()


def update_file_status(func):
    def wrapper(self, request):
        file_id, *args = request

        file_object = File.objects.get(id=file_id)

        next_status(file_object)
        return func(self, request)

    return wrapper


class UploadAbstractHandler(AbstractHandler):
    @staticmethod
    def get_file(file_id: str):
        return File.objects.get(id=file_id)

    @staticmethod
    def get_file_data(content: bytes, decode=False):
        if decode:
            content = base64.b64decode(content.decode(encoding="utf-8"))

        file_data = ContentFile(content)
        return file_data


class UploadHandler(UploadAbstractHandler):
    @staticmethod
    @shared_task
    def handle_file_upload(file_id: str, extension: str, content: bytes):
        file = UploadAbstractHandler.get_file(file_id)

        file_data = UploadAbstractHandler.get_file_data(content, decode=True)
        file.file.save(f"{file.id}.{extension}", file_data)

    @update_file_status
    def handle(self, request):
        file_id, extension, content = request

        self.handle_file_upload.delay(file_id, extension, content)
        return super().handle(request)


class ThumbnailHandler(UploadAbstractHandler):
    @staticmethod
    @shared_task
    def handle_thumbnail_generation(file_id: str, content: bytes):
        file = UploadAbstractHandler.get_file(file_id)
        file_data = UploadAbstractHandler.get_file_data(content, decode=True)

        try:
            value = Client(settings.GRPC_ADDR).generate_thumbnail(file_data.read())

            file_data = UploadAbstractHandler.get_file_data(value, decode=False)

            file.thumbnail.save(f"{file.id}_thumb.png", file_data)
        except Exception as e:
            print(e)

    @update_file_status
    def handle(self, request):
        file_id, _, content = request
        self.handle_thumbnail_generation.delay(file_id, content)
        return super().handle(request)

serializers.py #

class FileSerializer(serializers.ModelSerializer):
    class Meta:
        model = File
        fields = [
            "id",
            "created_at",
            "updated_at",
            "name",
            "group",
            "description",
            "tags",
            "status",
            "file",
            "size",
            "thumbnail",
        ]
        read_only_fields = [
            "id",
            "created_at",
            "updated_at",
            "status",
            "size",
            "thumbnail",
        ]

    def _get_file_ext(self, name: str):
        return name.split(".")[-1]

    def _get_valid_filename(self, name: str, file_obj):
        ext_from_file = self._get_file_ext(file_obj.name)

        if not name.lower().endswith(ext_from_file.lower()):
            name = f"{name}.{ext_from_file}"

        return name

    def validate(self, attrs):
        for key in attrs:
            if attrs[key] == "":
                attrs[key] = None
            if attrs[key] == [""]:
                attrs[key] = []

        attrs["name"] = self._get_valid_filename(attrs["name"], attrs["file"])

        return super().validate(attrs)

    def save(self, **kwargs):
        self.validated_data.pop("file")
        return super().save(**kwargs)
# 我在序列化器中移除了file字段，以通过Celery任务处理文件上传

你的问题核心是主文件上传任务完成后未将文件路径同步到数据库，而缩略图能正常同步可能是测试场景下的巧合（比如后续有其他操作触发了模型保存）。具体原因和修复方法如下：

核心原因 #

在UploadHandler的异步任务handle_file_upload中，你只完成了文件到存储系统的写入，但没有将模型实例的变更同步到数据库：

@staticmethod
@shared_task
def handle_file_upload(file_id: str, extension: str, content: bytes):
    file = UploadAbstractHandler.get_file(file_id)
    file_data = UploadAbstractHandler.get_file_data(content, decode=True)
    # 仅将文件保存到存储系统，更新了内存中file实例的file字段，但未同步到数据库
    file.file.save(f"{file.id}.{extension}", file_data)
    # 缺少关键的 file.save() 调用！

Django的FileField.save()方法只会处理文件的存储逻辑，并更新内存中的模型字段值，但不会自动将变更写入数据库，必须显式调用model.save()才能完成同步。

至于缩略图能正常同步到数据库，大概率是因为测试时的特殊场景（比如缩略图任务执行后，有其他业务操作触发了模型的save()，比如status字段的更新），但这不是可靠的默认行为。

修复步骤 #

在handle_file_upload任务末尾添加file.save()即可解决：

@staticmethod
@shared_task
def handle_file_upload(file_id: str, extension: str, content: bytes):
    file = UploadAbstractHandler.get_file(file_id)
    file_data = UploadAbstractHandler.get_file_data(content, decode=True)
    file.file.save(f"{file.id}.{extension}", file_data)
    # 新增：将文件路径同步到数据库
    file.save()

额外优化建议 #

1. 统一缩略图任务的保存逻辑：为了避免类似问题，建议在ThumbnailHandler.handle_thumbnail_generation中也显式添加file.save()，确保缩略图字段的变更能可靠同步：

@staticmethod
@shared_task
def handle_thumbnail_generation(file_id: str, content: bytes):
    file = UploadAbstractHandler.get_file(file_id)
    file_data = UploadAbstractHandler.get_file_data(content, decode=True)
    try:
        value = Client(settings.GRPC_ADDR).generate_thumbnail(file_data.read())
        file_data = UploadAbstractHandler.get_file_data(value, decode=False)
        file.thumbnail.save(f"{file.id}_thumb.png", file_data)
        # 新增：显式保存模型
        file.save()
    except Exception as e:
        print(e)

2. 防止并发修改问题：可以在获取模型实例时使用select_for_update，避免多任务并发修改导致的数据不一致：

@staticmethod
def get_file(file_id: str):
    return File.objects.select_for_update().get(id=file_id)

3. 完善错误日志：将print(e)替换为标准日志输出，方便排查问题：

import logging
logger = logging.getLogger(__name__)

# ...
except Exception as e:
    logger.error(f"生成缩略图失败：{str(e)}", exc_info=True)

内容的提问来源于stack exchange，提问作者youwish