向Celery任务传递音频文件出现JSON不可序列化错误的解决方案

你遇到的报错是因为Celery默认使用JSON作为任务参数的序列化器，而Django的UploadedFile文件对象属于不可序列化的复杂对象，直接作为参数传给delay()方法就会触发序列化失败错误。

核心原则是不要直接传递文件对象给Celery任务，只传递可序列化的基础类型参数（字符串、数字等），文件内容的读取由Celery Worker侧完成。

1. 调整Celery任务（tasks.py） #

任务接收可序列化的参数：新创建的演唱会ID、用户ID、上传文件的临时路径列表，把原视图中的音轨保存逻辑移到任务内。

import logging
from django.contrib.auth.models import User
from django.core.files import File
from Beyond_April_Base_Backend.celery import app
from .models import Concert, Song
import music_tag
import os

@app.task
def upload_songs(concert_id, user_id, temp_file_paths):
    try:
        user = User.objects.get(pk=user_id)
        concert = Concert.objects.get(pk=concert_id)
    except (User.DoesNotExist, Concert.DoesNotExist) as e:
        logging.warning(f"关联数据不存在：{str(e)}")
        return

    for index, temp_path in enumerate(temp_file_paths):
        # 读取临时文件元数据
        audio_metadata = music_tag.load_file(temp_path)
        song_title = str(audio_metadata['title'])
        # 打开临时文件创建Song实例
        with open(temp_path, 'rb') as f:
            song_file = File(f, name=os.path.basename(temp_path))
            audio_file_instance = Song(
                title=song_title,
                concert=concert,
                user=user,
                concert_order = index + 1,
                audio_file = song_file,
            )
            audio_file_instance.save()
    # 处理完后删除临时文件释放空间
    for temp_path in temp_file_paths:
        os.remove(temp_path)

2. 调整视图逻辑（views.py） #

视图侧先完成Concert实例的保存，再把上传的文件转存到持久化临时目录（避免Django请求结束后自动删除临时文件），将可序列化的参数传给Celery任务即可。

import os
from datetime import datetime
from django.conf import settings
from rest_framework import status, permissions
from rest_framework.response import Response
from rest_framework.views import APIView
from django.core.exceptions import ObjectDoesNotExist
from .models import Band, Concert
from .tasks import upload_songs

class ConcertUploadView(APIView):
    permission_classes = [permissions.IsAuthenticated]

    def post(self, request):
        track_files = request.FILES.getlist('files')
        current_user = self.request.user
        
        try:
            selected_band = Band.objects.get(name=request.data['band'])
        except ObjectDoesNotExist:
            print('band not received from form')
            selected_band = Band.objects.get(name='Band')
        venue_name = request.data['venue']
        concert_date_str = request.data['concertDate']
        concert_date_split = concert_date_str.split('(')[0]
        concert_date = datetime.strptime(concert_date_split, '%a %b %d %Y %H:%M:%S %Z%z ')
        concert_city = request.data['city']
        concert_state = request.data['state']
        concert_country = request.data['country']

        # 先保存Concert实例，拿到ID传给Celery
        new_concert = Concert(
            venue=venue_name,
            date=concert_date,
            city=concert_city,
            state=concert_state,
            country=concert_country,
            band=selected_band,
            user=current_user,
        )
        new_concert.save()

        # 把上传的文件存到自定义临时目录，避免请求结束后被删除
        temp_file_paths = []
        temp_dir = os.path.join(settings.MEDIA_ROOT, 'temp_uploads')
        os.makedirs(temp_dir, exist_ok=True)
        for track in track_files:
            temp_path = os.path.join(temp_dir, track.name)
            with open(temp_path, 'wb+') as f:
                for chunk in track.chunks():
                    f.write(chunk)
            temp_file_paths.append(temp_path)
        
        # 仅传递可序列化参数给Celery任务
        upload_songs.delay(new_concert.id, current_user.pk, temp_file_paths)

        return Response(status=status.HTTP_201_CREATED)

• 若你使用分布式Celery集群，需要将临时目录改为所有Worker节点都能访问的共享存储（如NAS、对象存储等），否则Worker无法读取到上传的文件。
• 不建议修改Celery的序列化器为pickle来直接传递文件对象，存在严重的安全风险。

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