问题标签 [celery-task]

使用模拟的返回值修补 Celery 任务调用，<Mock name='mock().get()' ...>而不是return_value由mock_task.get.return_value = "value". 但是，模拟任务在我的单元测试中正常运行。

这是我正在修补 Celery 任务的单元测试：

这是正在测试的功能：

最后一行加注TypeError: 'Mock' object has no attribute '__getitem__'

为什么我可以在我的单元测试中调用 mock_task.get() ，但调用它会foo返回 a<Mock ...>而不是预期的返回值？

我正在尝试运行以下代码：

但是此代码将 None 作为 request_id。请解释一下为什么我无法在侧芹菜任务类中读取 id

I am trying to get Celery work for awhile now. All my crontabs works just fine when I test it synchronously

but when I do

none of my crontabs work. I am not sure why

Note: I tried other schedule intervals as well like with time delta The same thing happens with that as well.

So I am fairly certain this is not a crontab thing but somehow related to the way I am starting celery multi.

Also, the worker turns on just fine since I can see it in Celery Flower but no tasks get executed.

我打算使用 celery 作为我项目的任务管理组件。它几乎具有我的项目所需的所有功能。我将有一组可以独立执行或按指定顺序执行的任务。在顺序任务中，如果中间任务之一失败，我希望能够执行清理/回滚。我想知道 celery 中是否有开箱即用的功能可以做同样的事情，或者是否有任何可用的解决方法。

我想要一个 celery 任务能够获取执行它的工作人员的名称，以用于记录目的。我需要从任务中处理这个问题，而不是直接查询代理。有没有办法做到这一点？如果这很重要，我将芹菜与 RabbitMQ 一起使用。

我想在我的用户注册 48 小时后向他们发送一封电子邮件。我如何使用 celery 实现这一点？如果我创建一个定期任务来发送电子邮件，我将不得不决定我想要执行该任务的具体时间。我不想每秒都运行一个 celery 任务来检查是否有任何电子邮件需要发送。

我正在为 Celery 结果添加后端，我在发送任务时遇到了问题，有些被接受，有些则不被接受。

已执行和未执行的任务都显示此日志输出：

那些被执行的然后显示这个输出：

虽然未执行的任务永远不会到达上述行。

我如何发送任务：

我的芹菜配置报告它是：

我的启动命令是：

此外，当我停止 celery 时，它会将从未被接受的任务从我的队列中拉出。然后当我重新启动时，它接受它们并执行它们。

非常感谢对此行为的任何帮助。我确定这与我的后端配置有关，但我很难隔离问题或其更正。谢谢！

总的来说，我对 celery 和 django 很陌生，所以请原谅我缺乏知识。我正在尝试运行测试以进行一些计算并等待测试完成，以便我可以确保它完成了正确的答案。

这是我所拥有的：

在 app/tests.py

在应用程序/tasks.py

即使在 celery 日志中表示任务已成功完成，状态也不会从挂起更改

我还在 mainapp/settings.py 中添加了 CELERY_IGNORE_RESULT = False，但这似乎没有任何作用。

运行 Celery 时收到 AssertionError。这个问题是持久的和可重复的，如果我运行一个包含约 700,000 个子任务的作业，那么问题会在某个时间点出现在作业中。但是，如果我运行一个只有 180 个子任务的作业，它就不会发生。

这是我的配置详细信息：

我的工人启动命令是：

我将我的应用程序作为一个任务运行，然后使用组合在一起的 celery 子任务启动子任务。

操作系统版本是 Mac OS X，版本 10.9.3。

这似乎与我在 Github 上看到的一些最近的 Celery 问题非常相似，但并不完全相同。谢谢！

根据与 celery 团队的对话，我也在 Github 上发布了这个问题：https ://github.com/celery/celery/issues/2086

我正在运行一个 Django 网站并且刚刚让 Celery 运行，但我遇到了令人困惑的错误。以下是代码的结构。

在测试.py 中：

在tasks.py中：

在 mainapp/settings.py 中：

芹菜调试控制台日志必须按列表/元组：

我得到的错误如上所述，AsyncResult(c.id).result: task args must be a list or tuple。这应该是一个简单的解决方案，但事实并非如此。当我把它列成这样的列表时：

然后它改变主意并告诉我AsyncResult(c.id).result: int() argument must be a string or a number, not 'list'

正如你可以想象的那样，我对可能是什么问题感到非常困惑。

编辑

任务.py

测试.py

function.app 打印