如何通过TaskResult获取Celery运行中任务的状态？

首先得明确：django_celery_results的TaskResult模型默认仅在任务进入最终状态（成功、失败、终止等）时才会写入/更新数据库，任务处于PENDING（等待中）或STARTED（运行中）状态时，数据库里大概率还没有对应的记录，这就是你拿到<QuerySet []>的原因。

下面给你两种可行的解决方案：

方案1：结合Celery原生AsyncResult获取实时状态（推荐） #

Celery的AsyncResult可以直接和消息中间件（如Redis、RabbitMQ）交互，实时获取任务的所有状态（包括PENDING、STARTED、SUCCESS等），如果需要把运行中状态持久化到TaskResult，可以手动同步：

from celery.result import AsyncResult
from django_celery_results.models import TaskResult
from django.http import HttpResponse

def test(request):
    # 触发任务并获取task_id
    task = test_task.delay()
    task_id = task.id
    
    # 用AsyncResult获取实时状态
    async_result = AsyncResult(task_id)
    current_status = async_result.status  # 可能的值：PENDING、STARTED、SUCCESS、FAILURE等
    
    # 手动同步状态到TaskResult（可选，如果你需要在数据库中保留运行中状态）
    task_result, created = TaskResult.objects.get_or_create(
        task_id=task_id,
        defaults={
            'task_name': test_task.name,
            'status': current_status,
            'date_created': async_result.date_created,
        }
    )
    if not created:
        task_result.status = current_status
        task_result.save()
    
    # 现在查询TaskResult就能拿到状态了
    status_record = TaskResult.objects.filter(task_id=task_id).first()
    print(f"当前任务状态：{status_record.status}")
    
    return HttpResponse(f"任务ID: {task_id}，当前状态: {current_status}")

方案2：配置Celery自动跟踪任务启动状态 #

通过开启Celery的task_track_started配置，让django_celery_results在任务开始执行时自动创建/更新TaskResult记录：

步骤1：修改Django settings.py配置 #

# settings.py
CELERY_TASK_TRACK_STARTED = True  # 开启任务启动跟踪
CELERY_RESULT_BACKEND = 'django-db'  # 确保使用django_celery_results的数据库后端
CELERY_RESULT_EXTENDED = True  # 可选，保存更多任务元数据

步骤2：获取状态的视图调整 #

开启配置后，当任务被worker拾取并开始执行（进入STARTED状态）时，TaskResult会自动创建记录。但注意：任务刚触发时还是PENDING状态，这时候数据库里依然没有记录，所以还是需要结合AsyncResult来获取这个阶段的状态：

from celery.result import AsyncResult
from django_celery_results.models import TaskResult
from django.http import HttpResponse

def test(request):
    task = test_task.delay()
    task_id = task.id
    
    # 先通过AsyncResult获取实时状态
    async_result = AsyncResult(task_id)
    current_status = async_result.status
    
    # 如果任务已经开始执行，TaskResult会存在
    if current_status == 'STARTED':
        task_result = TaskResult.objects.filter(task_id=task_id).first()
        print(f"数据库中任务状态：{task_result.status}")
    
    return HttpResponse(f"任务ID: {task_id}，当前状态: {current_status}")

注意事项 #

• PENDING状态：表示任务还未被worker拾取，此时数据库中不会有TaskResult记录，只能通过AsyncResult获取。
• STARTED状态：需要开启CELERY_TASK_TRACK_STARTED才会被跟踪，否则即使任务在运行，AsyncResult可能也只会返回PENDING。
• 如果你的Celery使用的是Redis/RabbitMQ作为消息中间件，AsyncResult的状态获取是实时且高效的，不需要依赖数据库。

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