Django post_save信号链式Celery任务报错：参数数量不匹配

问题背景 #

在Django的Game模型保存流程中，需求是先创建对应文件夹，再执行并行下载任务。使用Celery的chain和group编排任务时，触发错误：

TypeError: download_game() takes 2 positional arguments but 3 were given

原因是Celery链式任务默认会将前一个任务的返回结果（即使无返回值也会传入None）作为额外参数传递给下一个任务，导致目标任务的参数数量不匹配。

解决方案 #

使用Celery的**不可变签名（si()）**替代普通签名（s()），si()会忽略前序任务的返回值，不会自动传递额外参数，保证任务接收的参数数量完全符合定义。

修改后的代码 #

信号处理函数修改 #

@receiver(post_save, sender=Game)
def after_game_model_save(sender, instance, created, **kwargs):
    logger.info("after_game_model_save")
    task = None
    task_id = uuid4()
    # 改用si()创建不可变任务签名
    tasks_chain = chain(create_game_folder.si(instance.id))

    if created:
        # 同样使用si()避免传递前序结果
        tasks_chain |= download_game.si(instance.id, instance.download_link).set(
            task_id=str(task_id)
        )
    else:
        tasks_group = None
        if instance.tracker.has_changed("screenshots_urls"):
            tasks_group = group(
                [
                    download_game.si(instance.id, instance.download_link).set(
                        task_id=str(task_id)
                    ),
                    download_screenshots.si(instance.id),
                ]
            )
        if instance.tracker.has_changed("download_link"):
            tasks_group = group(
                [
                    download_game_update.si(instance.id, instance.download_link).set(
                        task_id=str(task_id)
                    ),
                    download_screenshots.si(instance.id),
                ]
            )
        if tasks_group:
            tasks_chain |= tasks_group
    tasks_chain.delay()
    try:
        task_obj = Task.objects.get(game=instance)
        task_obj.task_id = str(task_id)
        task_obj.save()
    except Task.DoesNotExist:
        Task.objects.create(game=instance, task_id=str(task_id))

任务定义保持不变 #

@shared_task()
def download_game(game_id, download_link):
    game_folder = Path(settings.GAMES_FOLDER / str(game_id))
    tmp_folder = game_folder / 'tmp'
    logger.info(f"Downloading {download_link} to {tmp_folder}")
    sleep(10)
    with open(tmp_folder/'test.txt', 'w') as fp:
        fp.write("##########################")
        fp.write("##########################")
        fp.write("##########################")
        fp.write("##########################")
        fp.write("##########################")
    
    logger.info(f"Downloaded {download_link} to {tmp_folder}")

原理说明 #

• s()：普通任务签名，会自动将前序任务的返回值作为第一个参数传递给当前任务，适用于需要依赖前序结果的场景。
• si()：不可变任务签名，完全忽略前序任务的返回值，仅使用显式传入的参数，适合无需依赖前序结果的链式任务编排。

这样修改后，链式任务会按顺序执行（先创建文件夹，再执行下载组/任务），同时不会传递额外参数，解决参数不匹配的问题。

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