在芹菜中使用共享_task时，如何在任务执行之前运行代码

Question

我要求我所有的芹菜任务都必须用特定的关键字参数调用。我想在执行任务之前检查并使用关键字的值。 例如，假设我有以下内容：

@shared_task
def my_task(*args, **kwargs):
    foo = kwargs.get('bar')  # -> I don't want to copy this to all my tasks
    # Do stuff here

如何创建一个名为my_special_shared_task的新装饰器，以便以下内容等同于上述：

@my_special_shared_task
def my_task(*args, **kwargs):
    # Do stuff here

Answer 1

那任务继承呢？
类似：

class BaseTask(celery.Task):
    foo = "some_value"


@app.task(base=BaseTask)
def my_task(*args, **kwargs):
    # Do stuff here
    print(self.foo)

在这里是文档。