如何通过Django信号触发Celery任务并传递模型实例？

直接传递Django模型实例给Celery任务会报错，因为Celery默认用JSON序列化参数，而模型实例无法被JSON序列化。正确的做法是传递实例的唯一标识（比如主键），在任务中重新查询实例。

修改信号函数 #

不要直接传instance，改为传递实例的主键，以及模型的标识（如果任务不固定处理某个模型的话）：

import uuid
def delete_content(sender, instance, **kwargs):
    task_id = uuid.uuid4()
    # 传递实例主键、模型名和所属app标签
    task = delete_libera_contents.apply_async(
        kwargs={
            "instance_pk": instance.pk,
            "model_name": sender.__name__,
            "app_label": sender._meta.app_label
        },
        task_id=task_id
    )
    task.wait(timeout=300, interval=2)

如果任务明确只处理MusicTracks模型，可以简化成直接传主键：

import uuid
def delete_content(sender, instance, **kwargs):
    task_id = uuid.uuid4()
    task = delete_libera_contents.apply_async(
        kwargs={"track_pk": instance.pk},
        task_id=task_id
    )
    task.wait(timeout=300, interval=2)

修改Celery任务 #

根据传递的标识重新查询模型实例，再执行原有逻辑：

通用版本（支持任意模型） #

from django.apps import apps
from celery import app

@app.task(name="Delete Libera Contents", queue='high_priority_tasks')
def delete_libera_contents(instance_pk, model_name, app_label, **kwargs):
    # 获取对应的模型类
    Model = apps.get_model(app_label=app_label, model_name=model_name)
    # 从数据库查询实例
    instance = Model.objects.get(pk=instance_pk)
    # 原有业务逻辑
    libera_backend = instance.file.libera_backend
    ...

针对MusicTracks的简化版本 #

from your_app.models import MusicTracks
from celery import app

@app.task(name="Delete Libera Contents", queue='high_priority_tasks')
def delete_libera_contents(track_pk, **kwargs):
    instance = MusicTracks.objects.get(pk=track_pk)
    libera_backend = instance.file.libera_backend
    ...

为什么不能直接传模型实例？ #

Celery任务参数需要跨进程（甚至跨机器）传输，默认用JSON序列化。Django模型实例包含数据库连接、内部状态等非序列化数据，无法被JSON处理。传递主键后在任务中重新查询，不仅解决了序列化问题，还能保证任务执行时拿到的是数据库中最新的实例数据。

注意：task.wait()会阻塞当前进程直到任务完成，如果是Web请求触发的信号，可能会拖慢响应速度，若无需同步等待结果，可去掉该调用。

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