Python 如何手动从 shell 运行 celery 定期任务？
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/12900023/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
How can I run a celery periodic task from the shell manually?
提问by Mridang Agarwalla
I'm using celery and django-celery. I have defined a periodic task that I'd like to test. Is it possible to run the periodic task from the shell manually so that I view the console output?
我正在使用芹菜和 django-celery。我已经定义了一个我想要测试的周期性任务。是否可以手动从 shell 运行周期性任务以便我查看控制台输出？
采纳答案by Platinum Azure
Have you tried just running the task from the Django shell? You can use the .applymethod of a task to ensure that it is run eagerly and locally.
您是否尝试过从 Django shell 运行任务？您可以使用.apply任务的方法来确保它在本地热切地运行。
Assuming the task is called my_taskin Django app myappin a taskssubmodule:
假设任务my_task在 Django 应用程序myapp中的tasks子模块中调用：
$ python manage.py shell
>>> from myapp.tasks import my_task
>>> eager_result = my_task.apply()
The result instance has the same API as the usual AsyncResulttype, except that the result is always evaluated eagerly and locally and the .apply()method will block until the task is run to completion.
结果实例与通常的AsyncResult类型具有相同的 API ，除了结果总是急切地在本地进行评估，并且该.apply()方法将阻塞，直到任务运行完成。
回答by darkphoenix
I think you'll need to open two shells: one for executing tasks from the Python/Django shell, and one for running celery worker(python manage.py celery worker). And as the previous answer said, you can run tasks using apply()or apply_async()
我认为您需要打开两个 shell：一个用于从 Python/Django shell 执行任务，另一个用于运行celery worker( python manage.py celery worker)。正如上一个答案所说，您可以使用apply()或运行任务apply_async()
I've edited the answer so you're not using a deprecated command.
我已经编辑了答案，因此您没有使用已弃用的命令。
回答by W.Perrin
If you means just trigger a task when the condition is not satisfied, for example, the periodic time not meet. You can do it in two steps.
如果你的意思是只在条件不满足时触发一个任务，例如周期时间不满足。您可以分两步完成。
1.Get your task id.
1.获取您的任务ID。
You can do it by typing.
您可以通过键入来完成。
celery inspect registered
You will see something like app.tasks.update_something.
If nothing, it is probably that celerywas not started. Just run it.
你会看到类似的东西app.tasks.update_something。如果没有，那可能celery是没有开始。运行它。
2.Run the task with celery call
2.运行任务 celery call
celery call app.tasks.update_something
For more details, just type
有关更多详细信息，只需键入
celery --help
celery inspect --help
celery call --help
