9

我正在尝试在 celery @task 装饰器之后应用一个装饰器，例如。

@send_email
@task
def any_function():

   print "inside the function"

我可以让它按照文档中推荐的方式工作，即将装饰器放在任务装饰器之前，但在这种情况下，我想在我的装饰器中访问任务实例。

@send_email 必须是类装饰器，这是我尝试但没有成功的方法：

class send_email(object):
    ''' wraps a Task celery class '''
    def __init__(self, obj):

        self.wrapped_obj = obj

        functools.update_wrapper(self, obj)

    def __call__(self, *args, **kwargs):

        print "call"

        return self.wrapped_obj.__call__(*args, **kwargs)

    def run(self, *args, **kwargs):

        print "run"

        return self.wrapped_obj.__call__(*args, **kwargs)

    def __getattr__(self, attr):

        if attr in self.__dict__:

            return getattr(self, attr)

        return getattr(self.wrapped_obj, attr)

我永远无法让调用或运行函数中的打印语句出现在工作程序或调用程序中。

我们如何在不使用基于类的任务定义的情况下装饰 celery 任务（因此装饰器将位于函数定义之上的 @task 之上。

谢谢你的帮助！

米格尔

4

2 回答 2

13

任务装饰器不返回一个类，它返回一个实例。

看来您的问题实际上应该是“我如何才能访问装饰器内部的任务”，而不是如何首先应用装饰器。

在即将发布的 3.1（开发版）中，您可以使用绑定任务来完成此操作：

def send_email(fun):
    @wraps(fun)
    def outer(self, *args, **kwargs):
        print('decorated and task is {0!r}'.format(self))
        return fun(self, *args, **kwargs)

    return outer

@task(bind=True)
@send_email
def any_function(self):
    print('inside the function')

对于以前的版本，您可以使用current_task：

from celery import current_task

def send_email(fun):

    @wraps(fun)
    def outer(*args, **kwargs):
        print('decorated and task is: %r' % (current_task, ))
        return fun(*args, **kwargs)

    return outer

@task
@send_email
def any_function():
    print('inside the function')
于 2013-10-16T12:00:35.513 回答
5

“之前”在视觉上看起来像“之后”。

例如，这个：

@decorator1
@decorator2
@decorator3
def func():
  pass

相当于：

def func():
  pass

func = decorator1(decorator2(decorator3(func)))

这意味着您必须先写@send_emailafter@task才能将其应用 before @task。例如：

@task
@send_email
def any_function():
    print "inside the function"
于 2013-10-16T09:06:37.460 回答