python - Celery 任务函数自定义属性

Question

我有一个看起来像这样的芹菜任务功能-

@task(base=MyBaseTask)
@my_custom_decorator 
def my_task(*args, **kwargs):
    my_task.ltc.some_func() #fails - attribute ltc doesn't exist on the object

my_custom_decorator 看起来像这样

def my_custom_decorator (f):
    from functools import wraps
    ltc = SomeClass()
    @wraps(f)
    def _inner(*args, **kwargs):
        ret_obj = None
        try:
            f.task_cache = ltc
            ret_obj = f(*args, **kwargs)
        except Exception, e:
            raise
        return ret_obj
    _inner.ltc = ltc
    return _inner

我看到这是因为调用来执行任务的实际可调用对象是 celery 任务类类型的对象。如何在此对象上保留我的属性“ltc”，以便可以从任务中访问它，如上图所示 - my_task.ltc.some_func()？

谢谢，

Answer 1

我认为一种简单的方法是ltc作为关键字参数引入..

@task(base=MyBaseTask)
@my_custom_decorator 
def my_task(*args, **kwargs):
    ltc = kwargs['ltc']
    ltc.some_func()

也许这样：

def my_custom_decorator (f):
    from functools import wraps
    ltc = SomeClass()
    @wraps(f)
    def _inner(*args, **kwargs):
        ret_obj = None
        try:
            f.task_cache = ltc
            kwargs['ltc'] = ltc
            ret_obj = f(*args, **kwargs)
        except Exception, e:
            raise
        return ret_obj
    _inner.ltc = ltc
    return _inner

我不知道是否有芹菜的任务方式可以做到这一点。希望这对您有所帮助。