5

我想通过手动事务管理运行 Django - Celery 任务，但注释似乎没有堆叠。

例如

def ping():
    print 'ping'
    pong.delay('arg')

@task(ignore_result=True)
@transaction.commit_manually()
def pong(arg):
    print 'pong: %s' % arg
    transaction.rollback()

结果是

TypeError: pong() got an unexpected keyword argument 'task_name'

而反向注释顺序导致

---> 22     pong.delay('arg')

AttributeError: 'function' object has no attribute 'delay'

这是有道理的，但我很难找到一个好的解决方法。Django 文档没有提到注释的替代方案，当我不需要时，我不想为每个 celery 任务创建一个类。

有任何想法吗？

4

2 回答 2

8

以前 Celery 有一些魔法，如果它接受了一组默认关键字参数，就会将它们传递给任务。

从 2.2 版开始，您可以禁用此行为，但最简单的方法是从以下位置导入task装饰celery.task器celery.decorators：

from celery.task import task

@task
@transaction.commit_manually
def t():
    pass

该decorators模块已弃用，将在 3.0 中完全删除，“魔术关键字参数”也是如此

注意：对于自定义任务类，您应该将accept_magic_kwargs属性设置为 False：

class MyTask(Task):
    accept_magic_kwargs = False

注意2：确保您的自定义装饰器使用 保留函数的名称functools.wraps，否则任务将以错误的名称结束。

于 2011-09-09T16:23:20.133 回答
6

class x(Task)任务装饰器从您的函数生成一个以run方法为目标的方法。建议你定义类并装饰方法。

未经测试，例如：

class pong(Task):
  ignore_result = True

  @transaction.commit_manually()
  def run(self,arg,**kwargs):
    print 'pong: %s' % arg
    transaction.rollback()
于 2011-09-08T15:30:37.933 回答