如何获取传递给Celery任务的参数？

Question

我在我的Web应用程序中使用celery与redis后端。 官方FAQ提供了获取任务结果的方式，只要有任务的ID就能获取。

result = my_task.AsyncResult(task_id)
result.get()

我可以轻松地在任务内部访问参数：

@app.task
def my_task(foo, bar, baz=None):
    kwargs = self.request.kwargs
    args = self.request.args

是否有一种方法可以从AsyncResult或其他地方仅通过任务ID获取args和kwargs？

- Sergey Belash

Answer 1

对于新版 Celery，您可以启用 "result_extended" 并从任务对象中获取 args：

app.conf.update(
   result_extended=True
)

然后，只需：

print(task.args)

- José Ibañez

---

FYI，Celery事件类型“task-received”表示事件字典具有“args”：但实际上，当我进入调试器并尝试获取“event ['args']”时，我得到的只是一个大约80个字符长的短字符串，并且仅列出了顶级字段与字符串中的“...”，因此我甚至无法执行“json.loads”或“eval()”。另外，FYI，“task-failed”没有“args”。-->结论使用“result_extended=True”。 - Trevor Boyd Smith

Answer 2

如果任务处于挂起状态或正在执行，您可以按照此处提到的方式查看任务的参数。

如果要获取失败任务的参数，您可以设置一个自定义基础任务并使用on_failure处理程序在任务失败时获取参数。

from celery import Task

class CustomBaseTask(Task):        

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print(args)

@app.task(base=CustomBaseTask)
def add(x, y):
    raise error

另一方面，如果您的任务成功执行，您将只获得结果，并且无法获取返回参数，因为这些信息已被代理商的工作人员消耗和处理。

- Chillar Anand

---

问题在于我需要这些参数不仅在任务成功/失败后，而且例如一周后，在redis中过期之前。现在我认为这是不可能的，因为celery不会将它们保存到后端。 - Sergey Belash

您可以使用任务ID组合将args显式保存在Redis中。 - Chillar Anand

只有在使用Redis时才需要这样做吗？其他后端呢？ - user528025

它也应该适用于其他后端。@user528025 - Chillar Anand