如何在Django中更新Celery任务的状态？

需求概述 #

需要在Django项目中实现多文本文件上传后，Celery处理任务的状态实时更新，并在前端表格中展示对应文件的任务状态。

步骤实现 #

1. 修正并完善Celery任务代码（tasks.py） #

原任务未接收文件参数，调整为接收文件ID，同时支持任务状态追踪：

from celery import shared_task
from .models import File

@shared_task(bind=True)
def text_analysis(self, file_id):
    # 获取目标文件对象
    file_obj = File.objects.get(id=file_id)
    try:
        # 更新任务为处理中状态
        self.update_state(state='PROGRESS', meta={'status': '正在提取文件数字...'})
        
        # 执行文本分析核心逻辑（示例）
        with file_obj.file.open() as f:
            content = f.read().decode('utf-8')
            numbers = [int(s) for s in content.split() if s.isdigit()]
            result = {'提取到的数字': numbers, '数量': len(numbers)}
        
        # 更新文件记录状态与结果
        file_obj.status = 'SUCCESS'
        file_obj.result = str(result)
        file_obj.save()
        return result
    except Exception as e:
        # 处理失败时更新状态
        file_obj.status = 'FAILURE'
        file_obj.save()
        self.update_state(state='FAILURE', meta={'error': str(e)})
        raise

2. 修改视图中的文件上传处理函数 #

为每个上传文件创建数据库记录，关联Celery任务ID：

from django.shortcuts import render
from .models import File
from .tasks import text_analysis

def text_analysis_view(request):
    if request.method == 'POST':
        text_files = request.FILES.getlist("files")
        for text_file in text_files:
            # 创建文件记录，初始状态为待处理
            file_obj = File.objects.create(
                user=request.user,
                file=text_file,
                status='PENDING'
            )
            # 启动Celery任务并保存任务ID
            task = text_analysis.delay(file_obj.id)
            file_obj.task_id = task.id
            file_obj.save()
    # 获取当前用户的所有文件记录
    files = File.objects.filter(user=request.user).order_by('-created_at')
    return render(request, "demo_demo.html", {"files": files})

3. 调整数据库模型（models.py） #

添加task_id和status字段，关联任务与记录状态：

from django.db import models
from django.contrib.auth.models import User

class File(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    file = models.FileField(upload_to='text_files/')
    created_at = models.DateTimeField(auto_now_add=True)
    task_id = models.CharField(max_length=255, blank=True, null=True)
    # 状态选项：*PENDING*（待处理）、*PROGRESS*（处理中）、*SUCCESS*（完成）、*FAILURE*（失败）
    status = models.CharField(max_length=20, default='PENDING')
    result = models.TextField(blank=True, null=True)

4. 更新HTML模板展示状态 #

修改模板显示文件信息，并添加前端轮询逻辑实时更新状态：

<main>
<form method="POST" enctype="multipart/form-data" id="texts_form">
{% csrf_token %}            
<input name="files" type="file" multiple>
<button type="submit">提交</button>
</form>                 
<div class="table-data">
<div class="order">     
<div class="head">
<h3>最近上传的文本文件</h3>
</div>                                      
<table>
<thead>
<tr>
<th>文件名</th>                               
<th>创建时间</th>
<th>状态</th>
</tr>
</thead>                        
<tbody>
{% for file in files %}
<tr data-task-id="{{ file.task_id }}" data-file-id="{{ file.id }}">            
<td>{{ file.file.name }}</td>                                                                           
<td>{{ file.created_at|date:"Y-m-d H:i:s" }}</td>
<td class="status-cell">{{ file.status }}</td>                           
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</main>

<script>
// 轮询更新任务状态
function updateTaskStatuses() {
    const rows = document.querySelectorAll('tr[data-task-id]');
    rows.forEach(row => {
        const taskId = row.getAttribute('data-task-id');
        const statusCell = row.querySelector('.status-cell');
        if (taskId && statusCell.textContent !== 'SUCCESS' && statusCell.textContent !== 'FAILURE') {
            fetch(`/task-status/${taskId}/`)
                .then(response => response.json())
                .then(data => {
                    statusCell.textContent = data.status;
                    // 处理中状态显示详细进度
                    if (data.status === 'PROGRESS') {
                        statusCell.textContent += `: ${data.meta.status}`;
                    }
                })
                .catch(error => console.error('更新状态失败:', error));
        }
    });
}

// 每3秒执行一次状态更新
setInterval(updateTaskStatuses, 3000);
</script>

5. 添加任务状态查询API视图 #

在views.py中新增查询任务状态的接口：

from django.http import JsonResponse
from celery.result import AsyncResult

def task_status(request, task_id):
    task_result = AsyncResult(task_id)
    if task_result.state == 'PENDING':
        response = {
            'status': task_result.state,
            'meta': '任务等待中...'
        }
    elif task_result.state == 'PROGRESS':
        response = {
            'status': task_result.state,
            'meta': task_result.info.get('status', '处理中...')
        }
    elif task_result.state == 'SUCCESS':
        response = {
            'status': task_result.state,
            'meta': '任务完成'
        }
    else:
        # 处理失败或其他异常状态
        response = {
            'status': task_result.state,
            'meta': task_result.info.get('error', '任务失败')
        }
    return JsonResponse(response)

6. 配置URL路由 #

在urls.py中添加状态查询路由：

from django.urls import path
from . import views

urlpatterns = [
    path('text-analysis/', views.text_analysis_view, name='text_analysis'),
    path('task-status/<str:task_id>/', views.task_status, name='task_status'),
]

已尝试但未解决的方案 #

• 基于Django的Celery结果后端方案
• FastAPI+Celery的前端状态更新方案

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