Django视图返回Celery异步图片处理任务结果的最佳方式

我正在开发一个Django应用，通过Celery任务处理图片。视图中的POST方法接收输入并触发该任务，任务内部会基于处理后的图片创建Image模型实例。我希望在同一视图中返回该模型实例的序列化响应，请问实现此响应的最佳方式是什么？前端是否应通过其他URL获取结果？如何通知应用该模型实例已就绪可提供服务？

# models.py
class Image(models.Model):
    name = models.CharField(max_length=50, blank=True)
    ...

# tasks.py
@shared_task
def long_running_function(user, data):

    image = Image.objects.create()
    ...
    return image # ?

# views.py
class RequestImage(APIView):

    def post(self, request):
        ...
        image = celery_task_example.delay(
            request.user.id, 
            request.data
        )
        # if image is ready
        serializer = ImageSerializer(image)
        return Response(serializer.data)

1. 同一视图直接返回实例的局限性 #

Celery任务是异步执行的，delay()方法返回的是任务ID（AsyncResult对象），而非直接的Image实例。你当前代码里直接用这个对象序列化完全行不通——任务还未执行完成时，实例可能尚未创建；就算创建了，AsyncResult也不是模型实例，无法被ImageSerializer序列化。

如果硬要在同一POST请求里返回实例，只能用task.get()同步等待任务完成，但这会让请求阻塞，完全失去Celery异步处理的意义，绝对不推荐。

2. 最佳实现方式：异步触发+结果查询/推送 #

方案一：前端轮询查询结果（通用简单）

这是异步任务场景下的标准方案，实现成本低：

• 后端修改：
  1. 给Image模型新增状态字段，标记图片处理进度：

     # models.py
     class Image(models.Model):
         name = models.CharField(max_length=50, blank=True)
         status = models.CharField(max_length=20, default='pending', choices=[
             ('pending', '处理中'),
             ('success', '完成'),
             ('failed', '失败')
         ])
         # 其他字段...
     

  2. 任务中更新状态：处理开始时保持pending，完成后改为success，失败则设为failed：

     # tasks.py
     @shared_task
     def long_running_function(user_id, data):
         user = User.objects.get(id=user_id)
         image = Image.objects.create(user=user, name=data.get('name'))
         try:
             # 图片处理逻辑...
             image.status = 'success'
             image.save()
             return image.id
         except Exception as e:
             image.status = 'failed'
             image.save()
             raise e
     

  3. 原POST视图只返回任务ID和结果查询地址：

     # views.py
     class RequestImage(APIView):
         def post(self, request):
             task = long_running_function.delay(request.user.id, request.data)
             return Response({
                 'task_id': task.id,
                 'result_url': reverse('image-status', kwargs={'task_id': task.id})
             }, status=202)
     

  4. 新增状态查询视图，返回实例状态或序列化数据：

     class ImageStatusView(APIView):
         def get(self, request, task_id):
             task = long_running_function.AsyncResult(task_id)
             if task.state == 'SUCCESS':
                 image_id = task.result
                 image = Image.objects.get(id=image_id)
                 serializer = ImageSerializer(image)
                 return Response(serializer.data)
             elif task.state == 'FAILURE':
                 return Response({'status': 'failed', 'error': str(task.info)}, status=500)
             else:
                 return Response({'status': 'pending'})
     

• 前端逻辑：拿到任务ID和查询地址后，每隔1-3秒发起GET请求查询状态，直到返回成功的实例数据或失败信息。

方案二：WebSocket实时推送（进阶实时）

如果需要更实时的结果通知，可以用Django Channels实现WebSocket：

• 前端建立WebSocket连接，等待后端推送；
• 任务处理完成时，后端主动通过WebSocket向前端发送序列化后的Image实例数据，无需前端轮询。

3. 是否需要通过其他URL获取结果 #

是的，必须通过单独的URL查询结果。原POST请求的作用是触发异步任务，不能阻塞等待任务完成，单独的状态查询接口是异步任务场景下的标准设计。

4. 如何通知实例就绪 #

• 轮询方案：通过Image模型的status字段标记状态，前端查询时判断是否为success；
• WebSocket方案：任务完成后，后端主动将实例数据推送给前端；
• 也可以结合Celery的任务状态回调，但本质还是需要前端监听状态变化，不如直接查询模型状态直观可控。

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