芹菜中@task装饰器后的装饰器 [英] Decorator after @task decorator in celery

问题描述

我正在尝试在celery @task装饰器之后应用装饰器，类似。

I'm trying to apply a decorator after the celery @task decorator, something like.

@send_email
@task
def any_function():

   print "inside the function"

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

I can get it to work in the way it is recommended in the docs, i.e. to put the decorator before the task decorator, but in this case I would like to access the task instance in my decorator.

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

The @send_email would have to be a class decorator, this is what I tried without success:

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)

我永远都无法在调用中获得打印语句，也无法将run函数功能显示在工作程序或调用方中。

I can never get the print statement in the call or run function function to appear in either the worker or the caller.

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

How can we decorate a celery Task, without resorting to class-based task definitions (so the decorator would be above @task above a function definition.

谢谢您的帮助！

Miguel

推荐答案

任务装饰器不会不是返回类，而是返回一个实例。

The task decorator does not return a class, it returns an instance.

似乎您的问题应该是我如何才能访问装饰器中的任务，而不是如何可以先应用装饰器。

It seems that your question should really be "How can I have access to the task inside the decorator" rather than how you can apply the decorator first.

在即将到来的3.1中（开发ve rsion），您可以使用绑定任务来完成此任务：

In the upcoming 3.1 (development version) you can use bound tasks to accomplish this:

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 ：

For previous versions you can use 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')

这篇关于芹菜中@task装饰器后的装饰器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！