如何重试所有因数据库错误而失败的芹菜任务?



我们的 django 代码库中有多个共享任务,它们从数据库获取数据并进行操作。是否可以包含一个芹菜配置,该配置可以重试"所有"失败的任务,而不是在每个单独的任务中添加"try-catch">

考虑使用autoretry_for选项,请参阅芹菜文档

上。
指定后不为每个任务提供选项可能会更方便。不幸的是,当时autoretry_for只能作为任务装饰器参数传递,但有一种功能方法可以做到这一点

from functools import partial
from celery import shared_task
shared_task = partial(shared_task, autoretry_for=(RuntimeError,))))

相关内容

  • 没有找到相关文章

最新更新