我们的 django 代码库中有多个共享任务,它们从数据库获取数据并进行操作。是否可以包含一个芹菜配置,该配置可以重试"所有"失败的任务,而不是在每个单独的任务中添加"try-catch">
考虑使用autoretry_for
选项,请参阅芹菜文档
上。
指定后不为每个任务提供选项可能会更方便。不幸的是,当时autoretry_for
只能作为任务装饰器参数传递,但有一种功能方法可以做到这一点
from functools import partial
from celery import shared_task
shared_task = partial(shared_task, autoretry_for=(RuntimeError,))))