3

我有一些非常冗长的任务功能,我想从我的tasks.py模块中分离出来,但仍然可以像from myapp.tasks import my_task. 我尝试为每个任务创建子模块,但随后我必须在 中插入一些技巧tasks/__init__.py,这看起来很hackish(并且需要为任务提供与模块相同的名称):

import pkgutil

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

3 回答 3

4

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


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

所以,在tasks.py

from django.core import management

@task()
def my_task():
    management.call_command('my_task')
于 2013-01-31T18:14:10.987 回答
2

我对 Django-Celery 不熟悉,但对 Celery 很熟悉,所以这更像是一个一般性的说明,可以回答你的问题......

Celery 任务在执行时由名称标识。默认情况下,任务被赋予其 Python 位置的名称,即。'%s.%s' % (my_task.__module__, my_task.__name__),或“myapp.tasks.function_name”。也就是说,您可以通过为 @task 装饰器提供“名称”kwarg 来覆盖任务名称。因此,如果您想覆盖任务名称:

# We're in myapp.tasks but the task will not be named myapp.tasks.my_task as expected
@task(name='some.overridden.name')
def my_task():
    print 'Something'
于 2013-01-31T00:02:13.993 回答
2

以下也有效:

  • 创建一个包tasks
  • tasks在包内的 python 模块中传播你的任务
  • 将模块导入到tasks/__init__.py

例子:

django_app
| tasks
   | __init__.py
   | filesystem_tasks.py
| admin.py
| url.py
  • tasks/filesystem_tasks.py可以包含这个:

    from __future__ import absolute_import, unicode_literals
    from celery import shared_task
    
    @shared_task
    def add(x, y):
        return x + y
    
  • tasks/__init__.py可以包含这个:

    from .filesystem_tasks import *
    
  • 运行 celery 表示任务,因为它们将tasks.py在同一级别中声明,例如。url.py.

于 2017-02-21T15:23:34.460 回答