链接芹菜任务方法错误

Chaining Celery Task Methods Error

我正在尝试使用 celery 将两个 task_methods 链接在一起,但出现以下错误:

错误:

>>> from proj.tasks import A
>>> a = A()
>>> s = (a.add.s(1,2) | a.show.s()).delay().get()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/python2.6/site-packages/celery/result.py", line 175, in get
    raise meta['result']
TypeError: show() takes exactly 2 arguments (1 given)

请注意,将 常规 (独立函数)celery 任务链接在一起时我不会收到此错误,仅 task_methods(class 函数)。我无法判断 self 对象是否未被传递,或者第一个任务的结果是否未被传递。

这是我的项目布局:

proj/__init__.py
    celery_app.py
    tasks.py

tasks.py:

from __future__ import absolute_import
from celery import current_app
from celery.contrib.methods import task_method
from proj.celery_app import app

class A:

    def __init__(self):
        self.something = 'something'

    @current_app.task(filter=task_method)
    def add(self,x, y):
        return x + y

    @current_app.task(filter=task_method)
    def show(self,s):
        print s
        return s

celery_app.py:

from __future__ import absolute_import
from celery import Celery

app = Celery('proj',
         broker='amqp://',
         backend='amqp://',
         include=['proj.tasks'])

app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=3600,
)

if __name__ == '__main__':
    app.start()

这是芹菜工人的错误:

[2015-04-15 19:57:52,338: ERROR/MainProcess] Task proj.tasks.show[e1e5bc12-6d36-46cd-beb7-fd92a0a5f5c2] raised unexpected: TypeError('show() takes exactly 2 arguments (1 given)',)
Traceback (most recent call last):
  File "/usr/lib/python2.6/site-packages/celery/app/trace.py", line 240, in trace_task
R = retval = fun(*args, **kwargs)
  File "/usr/lib/python2.6/site-packages/celery/app/trace.py", line 438, in __protected_call__
return self.run(*args, **kwargs)
TypeError: show() takes exactly 2 arguments (1 given)

有人成功将 task_methods 与 celery 链接起来了吗?谢谢!

edit:另外值得注意的是下面的代码成功的:

>>> from proj.tasks import A
>>> a = A()
>>> sum = a.add.s(1,2).delay().get()
>>> show = a.show.s(sum).delay().get()

此外,我知道有人可能会争辩说这些函数不需要在 class 中,但假装它们在。我使用简单的函数来帮助说明问题。

编辑

我找到了解决方法,但仍需要更好的解决方案:

首先修改tasks.py:

...
def show(s,self):
    print s
    return s
...

然后可以调用s = (a.add.s(1,1) | a.show.s(a) ).delay().get(),设置s为2。

有趣的是,调用 s = (a.add.s(1,1) | a.show.s(self=a) ).delay().get() 会返回以下错误:

TypeError: s() got multiple values for keyword argument 'self'

这并不理想,因为只有在链中才能调用 show 函数。

例如以下问题:

>>> a.show(s='test')
TypeError: show() got multiple values for keyword argument 's'

>>> a.show(s='test',self=a)
TypeError: __call__() got multiple values for keyword argument 'self'

根据 Celery 的创建者 Ask Solem Hoel 的说法,任务方法是 "failed experiment",将不再受支持。我想这回答了我的问题 - 目前无法完成。

Source