我有下面的项目树(用于测试目的),我试图了解芹菜是如何加载任务的
app
├── __init__.py
├── app.py
├── celery.py
└── my_tasks
├── __init__.py
└── tasks.py
celery.py
包含以下用于创建芹菜实例的代码:
from celery import Celery
app = Celery("app", backend="rpc://", broker="redis://localhost:6379/0")
app.autodiscover_tasks()
tasks.py
创建一个任务:
from app.celery import app
from time import sleep
@app.task
def run_tests_for_hash():
# task code here
和app.py
包含一个FastAPI应用程序,其中一个端点用于创建任务
from fastapi import FastAPI
from app.call_flow_tasks.tasks import run_tests_for_hash
app = FastAPI(title="Testing Studio runners")
@app.post("/create_task")
def create_task():
results = run_tests_for_hash.delay()
return {"task_id": results.id, "status": results.status}
但尝试从命令行运行芹菜工人时,我遇到一个错误:
$celery -A app worker
Traceback (most recent call last):
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/bin/celery", line 8, in <module>
sys.exit(main())
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/__main__.py", line 16, in main
_main()
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/celery.py", line 322, in main
cmd.execute_from_commandline(argv)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/celery.py", line 495, in execute_from_commandline
super(CeleryCommand, self).execute_from_commandline(argv)))
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/base.py", line 305, in execute_from_commandline
return self.handle_argv(self.prog_name, argv[1:])
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/celery.py", line 487, in handle_argv
return self.execute(command, argv)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/celery.py", line 419, in execute
).run_from_argv(self.prog_name, argv[1:], command=argv[0])
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/worker.py", line 223, in run_from_argv
return self(*args, **options)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/base.py", line 253, in __call__
ret = self.run(*args, **kwargs)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/bin/worker.py", line 258, in run
**kwargs)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/worker/worker.py", line 96, in __init__
self.app.loader.init_worker()
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/loaders/base.py", line 114, in init_worker
self.import_default_modules()
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/loaders/base.py", line 108, in import_default_modules
raise response
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/utils/dispatch/signal.py", line 288, in send
response = receiver(signal=self, sender=sender, **named)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/vine/promises.py", line 170, in __call__
return self.throw()
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/vine/promises.py", line 167, in __call__
retval = fun(*final_args, **final_kwargs)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/app/base.py", line 695, in _autodiscover_tasks
return self._autodiscover_tasks_from_fixups(related_name)
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/app/base.py", line 705, in _autodiscover_tasks_from_fixups
pkg for fixup in self._fixups
File "/Users/avlachopoulos/.venvs/test-runner-JK5CNUCV/lib/python3.7/site-packages/celery/app/base.py", line 706, in <listcomp>
for pkg in fixup.autodiscover_tasks()
AttributeError: 'NoneType' object has no attribute 'autodiscover_tasks'
命令在与app
python包相同的级别上执行。如果我删除autodiscover
,它会正常工作,并且工作进程会正常加载。有关芹菜如何自动发现任务以及如何从不同模块加载任务的任何帮助
您可以组合两种类型的自动发现:
芹菜只有一个
BUILTIN_FIXUPS = {'celery.fixups.django:fixup'}
,看它是code to understand来自芹菜文档:
https://docs.celeryproject.org/en/stable/reference/celery.html#celery.Celery.autodiscover_tasks
您应该为
autodiscover_tasks
方法提供一个包名列表,如果您不提供,它将被委托给修复程序(例如:Django-只要您正确配置这些设置)最简单的方法是将任务模块添加到包列表下:
app.autodiscover_tasks(packages=['my_tasks'])
然后您就可以在命令行上运行worker
希望有帮助
如果您的应用程序更大,则可能会:
在worker.py中
相关问题 更多 >
编程相关推荐