如何在代码中检测Celery任务完成状态并更新数据库记录的时机与实现方案

兄弟，你的问题其实是Celery任务状态同步到数据库的典型场景，我给你三个实用的方案，从优雅到简单，你可以根据自己的业务场景灵活选择：

方案一：用Celery任务信号（最推荐，自动化程度最高） #

Celery本身提供了信号机制，可以在任务的不同生命周期节点（比如成功、失败、开始执行）自动触发回调函数。这是最优雅的方式，不需要改动业务逻辑，也不需要依赖前端轮询，任务状态变化时会自动同步到数据库。

实现步骤： #

在你的logic.py里添加信号处理函数，监听任务的成功/失败事件：

from celery import shared_task, signals
from celery.result import AsyncResult
from .models import CeleryJobResult  # 替换成你的模型实际导入路径

# 监听任务成功的信号
@signals.task_success.connect(sender=run_create_or_update_google_creative)
def update_job_on_success(sender=None, **kwargs):
    task_id = sender.request.id
    try:
        # 找到对应的数据库记录
        job = CeleryJobResult.objects.get(job_id=task_id)
        # 更新为Celery返回的成功状态
        job.status = AsyncResult(task_id).status
        job.save()
    except CeleryJobResult.DoesNotExist:
        # 找不到记录时可以打日志或者忽略，根据需求调整
        pass

# 监听任务失败的信号（可选，用来记录错误信息）
@signals.task_failure.connect(sender=run_create_or_update_google_creative)
def update_job_on_failure(sender=None, exception=None, **kwargs):
    task_id = sender.request.id
    try:
        job = CeleryJobResult.objects.get(job_id=task_id)
        job.status = 'FAILURE'
        job.error_message = str(exception)  # 可以把错误信息存到数据库，方便排查
        job.save()
    except CeleryJobResult.DoesNotExist:
        pass

时机说明：当run_create_or_update_google_creative任务成功完成时，update_job_on_success会自动执行；任务抛出异常失败时，update_job_on_failure自动执行，完全不需要手动触发。

方案二：在任务内部更新状态（适合简单场景） #

如果你的任务逻辑比较简单，不想配置信号，可以直接在任务的业务逻辑结尾（以及异常处理块）更新数据库状态。

实现步骤： #

修改create_or_update_google_creative函数，添加状态更新逻辑：

def create_or_update_google_creative():
    try:
        # 你的原有业务逻辑
        # do some logic...
        
        # 任务成功后，更新数据库状态
        task_id = run_create_or_update_google_creative.request.id
        job = CeleryJobResult.objects.get(job_id=task_id)
        job.status = 'SUCCESS'
        job.save()
    except Exception as e:
        # 任务失败时，更新为失败状态
        task_id = run_create_or_update_google_creative.request.id
        try:
            job = CeleryJobResult.objects.get(job_id=task_id)
            job.status = 'FAILURE'
            job.error_message = str(e)
            job.save()
        except CeleryJobResult.DoesNotExist:
            pass
        raise  # 必须重新抛出异常，让Celery记录任务失败状态

时机说明：任务正常执行完业务逻辑后，会立即更新状态；如果抛出异常，会进入except块更新失败状态。

方案三：前端轮询时同步状态（适合依赖前端交互的场景） #

如果你希望和前端的轮询流程结合，每次前端查询状态时，后端主动去Celery拉取最新状态并同步到数据库，这也是可行的。

实现步骤： #

把你提到的get_task_status改造成接口（以Django视图为例）：

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

def get_task_status(request, task_id):
    try:
        job = CeleryJobResult.objects.get(job_id=task_id)
        celery_task = AsyncResult(task_id)
        
        # 如果Celery的最新状态和数据库不一致，就更新数据库
        if celery_task.status != job.status:
            job.status = celery_task.status
            if celery_task.status == 'FAILURE':
                # 把Celery返回的错误信息存下来
                job.error_message = str(celery_task.info)
            job.save()
        
        # 返回最新状态给前端
        return JsonResponse({
            'task_id': task_id,
            'status': job.status,
            'error_message': job.error_message if hasattr(job, 'error_message') else None
        })
    except CeleryJobResult.DoesNotExist:
        return JsonResponse({'error': '任务不存在'}, status=404)

时机说明：每次前端调用这个接口轮询时，后端会检查并同步最新状态，相当于把状态更新的时机绑定到前端的查询动作上。

方案对比 #

备注：内容来源于stack exchange，提问作者Jekson