python - 我如何将一个大的 django-celery tasks.py 模块拆分成更小的 block ？

Question

我有一些非常冗长的任务函数，我想从我的 tasks.py 模块中分离出来，但仍然可以调用 from myapp.tasks import my_task。我尝试为每个任务创建子模块，但随后我不得不在 tasks/__init__.py 中插入一些技巧，这看起来很老套(并且需要为任务提供与模块相同的名称):

import pkgutil

for module_loader, name, ispkg in pkgutil.iter_modules(__path__):
    __import__('myapp.tasks.%s' % name)
    locals().update({name: getattr(locals()[name], name)})

Answer 1

这个答案是关于管理命令而不是 Celery 后台任务意义上的任务。

---

我认为最直接的解决方案是首先将任务编写为 Django 管理命令，然后从 tasks.py 中调用它们。通过这种方式，您可以将您的任务分解为每个文件一个任务，并获得能够从命令行任意调用您的任务的额外好处。

因此，在 tasks.py 中:

from django.core import management

@task()
def my_task():
    management.call_command('my_task')